/* * Licensed to the Apache Software Foundation (ASF) under one * or more contributor license agreements. See the NOTICE file * distributed with this work for additional information * regarding copyright ownership. The ASF licenses this file * to you under the Apache License, Version 2.0 (the * "License"); you may not use this file except in compliance * with the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, * software distributed under the License is distributed on an * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY * KIND, either express or implied. See the License for the * specific language governing permissions and limitations * under the License. */ package thrift import ( "bytes" "context" "encoding/binary" "fmt" "io" "math" ) type TBinaryProtocol struct { trans TRichTransport origTransport TTransport cfg *TConfiguration buffer [64]byte } type TBinaryProtocolFactory struct { cfg *TConfiguration } // Deprecated: Use NewTBinaryProtocolConf instead. func NewTBinaryProtocolTransport(t TTransport) *TBinaryProtocol { return NewTBinaryProtocolConf(t, &TConfiguration{ noPropagation: true, }) } // Deprecated: Use NewTBinaryProtocolConf instead. func NewTBinaryProtocol(t TTransport, strictRead, strictWrite bool) *TBinaryProtocol { return NewTBinaryProtocolConf(t, &TConfiguration{ TBinaryStrictRead: &strictRead, TBinaryStrictWrite: &strictWrite, noPropagation: true, }) } func NewTBinaryProtocolConf(t TTransport, conf *TConfiguration) *TBinaryProtocol { PropagateTConfiguration(t, conf) p := &TBinaryProtocol{ origTransport: t, cfg: conf, } if et, ok := t.(TRichTransport); ok { p.trans = et } else { p.trans = NewTRichTransport(t) } return p } // Deprecated: Use NewTBinaryProtocolFactoryConf instead. func NewTBinaryProtocolFactoryDefault() *TBinaryProtocolFactory { return NewTBinaryProtocolFactoryConf(&TConfiguration{ noPropagation: true, }) } // Deprecated: Use NewTBinaryProtocolFactoryConf instead. func NewTBinaryProtocolFactory(strictRead, strictWrite bool) *TBinaryProtocolFactory { return NewTBinaryProtocolFactoryConf(&TConfiguration{ TBinaryStrictRead: &strictRead, TBinaryStrictWrite: &strictWrite, noPropagation: true, }) } func NewTBinaryProtocolFactoryConf(conf *TConfiguration) *TBinaryProtocolFactory { return &TBinaryProtocolFactory{ cfg: conf, } } func (p *TBinaryProtocolFactory) GetProtocol(t TTransport) TProtocol { return NewTBinaryProtocolConf(t, p.cfg) } func (p *TBinaryProtocolFactory) SetTConfiguration(conf *TConfiguration) { p.cfg = conf } /** * Writing Methods */ func (p *TBinaryProtocol) WriteMessageBegin(ctx context.Context, name string, typeId TMessageType, seqId int32) error { if p.cfg.GetTBinaryStrictWrite() { version := uint32(VERSION_1) | uint32(typeId) e := p.WriteI32(ctx, int32(version)) if e != nil { return e } e = p.WriteString(ctx, name) if e != nil { return e } e = p.WriteI32(ctx, seqId) return e } else { e := p.WriteString(ctx, name) if e != nil { return e } e = p.WriteByte(ctx, int8(typeId)) if e != nil { return e } e = p.WriteI32(ctx, seqId) return e } return nil } func (p *TBinaryProtocol) WriteMessageEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) WriteStructBegin(ctx context.Context, name string) error { return nil } func (p *TBinaryProtocol) WriteStructEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) WriteFieldBegin(ctx context.Context, name string, typeId TType, id int16) error { e := p.WriteByte(ctx, int8(typeId)) if e != nil { return e } e = p.WriteI16(ctx, id) return e } func (p *TBinaryProtocol) WriteFieldEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) WriteFieldStop(ctx context.Context) error { e := p.WriteByte(ctx, STOP) return e } func (p *TBinaryProtocol) WriteMapBegin(ctx context.Context, keyType TType, valueType TType, size int) error { e := p.WriteByte(ctx, int8(keyType)) if e != nil { return e } e = p.WriteByte(ctx, int8(valueType)) if e != nil { return e } e = p.WriteI32(ctx, int32(size)) return e } func (p *TBinaryProtocol) WriteMapEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) WriteListBegin(ctx context.Context, elemType TType, size int) error { e := p.WriteByte(ctx, int8(elemType)) if e != nil { return e } e = p.WriteI32(ctx, int32(size)) return e } func (p *TBinaryProtocol) WriteListEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) WriteSetBegin(ctx context.Context, elemType TType, size int) error { e := p.WriteByte(ctx, int8(elemType)) if e != nil { return e } e = p.WriteI32(ctx, int32(size)) return e } func (p *TBinaryProtocol) WriteSetEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) WriteBool(ctx context.Context, value bool) error { if value { return p.WriteByte(ctx, 1) } return p.WriteByte(ctx, 0) } func (p *TBinaryProtocol) WriteByte(ctx context.Context, value int8) error { e := p.trans.WriteByte(byte(value)) return NewTProtocolException(e) } func (p *TBinaryProtocol) WriteI16(ctx context.Context, value int16) error { v := p.buffer[0:2] binary.BigEndian.PutUint16(v, uint16(value)) _, e := p.trans.Write(v) return NewTProtocolException(e) } func (p *TBinaryProtocol) WriteI32(ctx context.Context, value int32) error { v := p.buffer[0:4] binary.BigEndian.PutUint32(v, uint32(value)) _, e := p.trans.Write(v) return NewTProtocolException(e) } func (p *TBinaryProtocol) WriteI64(ctx context.Context, value int64) error { v := p.buffer[0:8] binary.BigEndian.PutUint64(v, uint64(value)) _, err := p.trans.Write(v) return NewTProtocolException(err) } func (p *TBinaryProtocol) WriteDouble(ctx context.Context, value float64) error { return p.WriteI64(ctx, int64(math.Float64bits(value))) } func (p *TBinaryProtocol) WriteString(ctx context.Context, value string) error { e := p.WriteI32(ctx, int32(len(value))) if e != nil { return e } _, err := p.trans.WriteString(value) return NewTProtocolException(err) } func (p *TBinaryProtocol) WriteBinary(ctx context.Context, value []byte) error { e := p.WriteI32(ctx, int32(len(value))) if e != nil { return e } _, err := p.trans.Write(value) return NewTProtocolException(err) } /** * Reading methods */ func (p *TBinaryProtocol) ReadMessageBegin(ctx context.Context) (name string, typeId TMessageType, seqId int32, err error) { size, e := p.ReadI32(ctx) if e != nil { return "", typeId, 0, NewTProtocolException(e) } if size < 0 { typeId = TMessageType(size & 0x0ff) version := int64(int64(size) & VERSION_MASK) if version != VERSION_1 { return name, typeId, seqId, NewTProtocolExceptionWithType(BAD_VERSION, fmt.Errorf("Bad version in ReadMessageBegin")) } name, e = p.ReadString(ctx) if e != nil { return name, typeId, seqId, NewTProtocolException(e) } seqId, e = p.ReadI32(ctx) if e != nil { return name, typeId, seqId, NewTProtocolException(e) } return name, typeId, seqId, nil } if p.cfg.GetTBinaryStrictRead() { return name, typeId, seqId, NewTProtocolExceptionWithType(BAD_VERSION, fmt.Errorf("Missing version in ReadMessageBegin")) } name, e2 := p.readStringBody(size) if e2 != nil { return name, typeId, seqId, e2 } b, e3 := p.ReadByte(ctx) if e3 != nil { return name, typeId, seqId, e3 } typeId = TMessageType(b) seqId, e4 := p.ReadI32(ctx) if e4 != nil { return name, typeId, seqId, e4 } return name, typeId, seqId, nil } func (p *TBinaryProtocol) ReadMessageEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) ReadStructBegin(ctx context.Context) (name string, err error) { return } func (p *TBinaryProtocol) ReadStructEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) ReadFieldBegin(ctx context.Context) (name string, typeId TType, seqId int16, err error) { t, err := p.ReadByte(ctx) typeId = TType(t) if err != nil { return name, typeId, seqId, err } if t != STOP { seqId, err = p.ReadI16(ctx) } return name, typeId, seqId, err } func (p *TBinaryProtocol) ReadFieldEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) ReadMapBegin(ctx context.Context) (kType, vType TType, size int, err error) { k, e := p.ReadByte(ctx) if e != nil { err = NewTProtocolException(e) return } kType = TType(k) v, e := p.ReadByte(ctx) if e != nil { err = NewTProtocolException(e) return } vType = TType(v) size32, e := p.ReadI32(ctx) if e != nil { err = NewTProtocolException(e) return } err = checkSizeForProtocol(size32, p.cfg) if err != nil { return } size = int(size32) return kType, vType, size, nil } func (p *TBinaryProtocol) ReadMapEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) ReadListBegin(ctx context.Context) (elemType TType, size int, err error) { b, e := p.ReadByte(ctx) if e != nil { err = NewTProtocolException(e) return } elemType = TType(b) size32, e := p.ReadI32(ctx) if e != nil { err = NewTProtocolException(e) return } err = checkSizeForProtocol(size32, p.cfg) if err != nil { return } size = int(size32) return } func (p *TBinaryProtocol) ReadListEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) ReadSetBegin(ctx context.Context) (elemType TType, size int, err error) { b, e := p.ReadByte(ctx) if e != nil { err = NewTProtocolException(e) return } elemType = TType(b) size32, e := p.ReadI32(ctx) if e != nil { err = NewTProtocolException(e) return } err = checkSizeForProtocol(size32, p.cfg) if err != nil { return } size = int(size32) return elemType, size, nil } func (p *TBinaryProtocol) ReadSetEnd(ctx context.Context) error { return nil } func (p *TBinaryProtocol) ReadBool(ctx context.Context) (bool, error) { b, e := p.ReadByte(ctx) v := true if b != 1 { v = false } return v, e } func (p *TBinaryProtocol) ReadByte(ctx context.Context) (int8, error) { v, err := p.trans.ReadByte() return int8(v), err } func (p *TBinaryProtocol) ReadI16(ctx context.Context) (value int16, err error) { buf := p.buffer[0:2] err = p.readAll(ctx, buf) value = int16(binary.BigEndian.Uint16(buf)) return value, err } func (p *TBinaryProtocol) ReadI32(ctx context.Context) (value int32, err error) { buf := p.buffer[0:4] err = p.readAll(ctx, buf) value = int32(binary.BigEndian.Uint32(buf)) return value, err } func (p *TBinaryProtocol) ReadI64(ctx context.Context) (value int64, err error) { buf := p.buffer[0:8] err = p.readAll(ctx, buf) value = int64(binary.BigEndian.Uint64(buf)) return value, err } func (p *TBinaryProtocol) ReadDouble(ctx context.Context) (value float64, err error) { buf := p.buffer[0:8] err = p.readAll(ctx, buf) value = math.Float64frombits(binary.BigEndian.Uint64(buf)) return value, err } func (p *TBinaryProtocol) ReadString(ctx context.Context) (value string, err error) { size, e := p.ReadI32(ctx) if e != nil { return "", e } err = checkSizeForProtocol(size, p.cfg) if err != nil { return } if size == 0 { return "", nil } if size < int32(len(p.buffer)) { // Avoid allocation on small reads buf := p.buffer[:size] read, e := io.ReadFull(p.trans, buf) return string(buf[:read]), NewTProtocolException(e) } return p.readStringBody(size) } func (p *TBinaryProtocol) ReadBinary(ctx context.Context) ([]byte, error) { size, e := p.ReadI32(ctx) if e != nil { return nil, e } if err := checkSizeForProtocol(size, p.cfg); err != nil { return nil, err } buf, err := safeReadBytes(size, p.trans) return buf, NewTProtocolException(err) } func (p *TBinaryProtocol) Flush(ctx context.Context) (err error) { return NewTProtocolException(p.trans.Flush(ctx)) } func (p *TBinaryProtocol) Skip(ctx context.Context, fieldType TType) (err error) { return SkipDefaultDepth(ctx, p, fieldType) } func (p *TBinaryProtocol) Transport() TTransport { return p.origTransport } func (p *TBinaryProtocol) readAll(ctx context.Context, buf []byte) (err error) { var read int _, deadlineSet := ctx.Deadline() for { read, err = io.ReadFull(p.trans, buf) if deadlineSet && read == 0 && isTimeoutError(err) && ctx.Err() == nil { // This is I/O timeout without anything read, // and we still have time left, keep retrying. continue } // For anything else, don't retry break } return NewTProtocolException(err) } func (p *TBinaryProtocol) readStringBody(size int32) (value string, err error) { buf, err := safeReadBytes(size, p.trans) return string(buf), NewTProtocolException(err) } func (p *TBinaryProtocol) SetTConfiguration(conf *TConfiguration) { PropagateTConfiguration(p.trans, conf) PropagateTConfiguration(p.origTransport, conf) p.cfg = conf } var ( _ TConfigurationSetter = (*TBinaryProtocolFactory)(nil) _ TConfigurationSetter = (*TBinaryProtocol)(nil) ) // This function is shared between TBinaryProtocol and TCompactProtocol. // // It tries to read size bytes from trans, in a way that prevents large // allocations when size is insanely large (mostly caused by malformed message). func safeReadBytes(size int32, trans io.Reader) ([]byte, error) { if size < 0 { return nil, nil } buf := new(bytes.Buffer) _, err := io.CopyN(buf, trans, int64(size)) return buf.Bytes(), err }