func readStatusCodeAndSessionId(dbc *DBClient) error {
	status, err := rw.ReadByte(dbc.conx)
	if err != nil {
		return oerror.NewTrace(err)
	}

	sessionId, err := rw.ReadInt(dbc.conx)
	if err != nil {
		return oerror.NewTrace(err)
	}
	if sessionId != dbc.sessionId {
		// FIXME: use of fmt.Errorf is an anti-pattern
		return fmt.Errorf("sessionId from server (%v) does not match client sessionId (%v)",
			sessionId, dbc.sessionId)
	}

	if status == RESPONSE_STATUS_ERROR {
		serverException, err := rw.ReadErrorResponse(dbc.conx)
		if err != nil {
			return oerror.NewTrace(err)
		}
		return serverException
	}

	return nil
}

//
// NewDBClient creates a new DBClient after contacting the OrientDB server
// specified in the ClientOptions and validating that the server and client
// speak the same binary protocol version.
// The DBClient returned is ready to make calls to the OrientDB but has not
// yet established a database session or a session with the OrientDB server.
// After this, the user needs to call either OpenDatabase or CreateServerSession.
//
func NewDBClient(opts ClientOptions) (*DBClient, error) {
	// binary port range is: 2424-2430
	if opts.ServerHost == "" {
		opts.ServerHost = "127.0.0.1"
	}
	if opts.ServerPort == "" {
		opts.ServerPort = "2424"
	}
	hostport := fmt.Sprintf("%s:%s", opts.ServerHost, opts.ServerPort)
	conx, err := net.Dial("tcp", hostport)
	if err != nil {
		fmt.Fprintf(os.Stderr, "WARN: %v\n", err)
		return nil, oerror.NewTrace(err)
	}

	// after connecting the OrientDB server sends back 2 bytes - its binary protocol version
	readbuf := make([]byte, 2)
	n, err := conx.Read(readbuf)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	buf := new(bytes.Buffer)
	_, err = buf.Write(readbuf[0:n])
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	var (
		svrProtocolNum int16
		serdeV0        ORecordSerializer
		serializerType string
	)
	binary.Read(buf, binary.BigEndian, &svrProtocolNum)
	if svrProtocolNum < MinSupportedBinaryProtocolVersion {
		return nil, ErrUnsupportedVersion{serverVersion: svrProtocolNum}
	} else if svrProtocolNum > MaxSupportedBinaryProtocolVersion {
		return nil, ErrUnsupportedVersion{serverVersion: svrProtocolNum}
	}

	serializerType = BinarySerialization
	serdeV0 = &ORecordSerializerV0{}
	if svrProtocolNum < MinBinarySerializerVersion {
		serializerType = CsvSerialization
		panic(fmt.Sprintf("Server Binary Protocol Version (%v) is less than the Min Binary Serializer Version supported by this driver (%v)",
			svrProtocolNum, MinBinarySerializerVersion))
	}

	dbc := &DBClient{
		conx:                  conx,
		buf:                   new(bytes.Buffer),
		serializationType:     serializerType,
		binaryProtocolVersion: svrProtocolNum,
		serializationVersion:  byte(0), // default is 0 // TODO: need to detect if server is using a higher version
		sessionId:             NoSessionID,
		RecordSerDes:          []ORecordSerializer{serdeV0},
	}

	return dbc, nil
}

//
// Returns map of string keys to *oschema.OLink
//
func (serde ORecordSerializerV0) readLinkMap(buf io.Reader) (map[string]*oschema.OLink, error) {
	nentries, err := varint.ReadVarIntAndDecode32(buf)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	linkMap := make(map[string]*oschema.OLink)

	for i := 0; i < int(nentries); i++ {
		/* ---[ read map key ]--- */
		datatype, err := rw.ReadByte(buf)
		if err != nil {
			return nil, oerror.NewTrace(err)
		}
		if datatype != byte(oschema.STRING) {
			// FIXME: even though all keys are currently strings, it would be easy to allow other types
			//        using serde.readDataValue(dbc, buf, serde)
			return nil, fmt.Errorf("readLinkMap: datatype for key is NOT string but type: %v", datatype)
		}

		mapkey, err := varint.ReadString(buf)
		if err != nil {
			return nil, oerror.NewTrace(err)
		}

		/* ---[ read map value (always a RID) ]--- */
		mapval, err := serde.readLink(buf)
		if err != nil {
			return nil, oerror.NewTrace(err)
		}
		linkMap[mapkey] = mapval
	}

	return linkMap, nil
}

//
// Serialize takes an ODocument and serializes it to bytes in accordance
// with the OrientDB binary serialization spec and writes them to the
// bytes.Buffer passed in.
//
func (serde ORecordSerializerV0) Serialize(dbc *DBClient, doc *oschema.ODocument) ([]byte, error) {
	// temporarily set state for the duration of this Serialize call
	// dbc is allowed to be nil for reentrant (recursive) calls -- in which
	// case serde.dbc should already be set (not-nil)
	if dbc != nil {
		if serde.dbc != nil {
			return nil, errors.New("Attempted to set dbc again in Serialize when it is already set")
		}
		serde.dbc = dbc
		defer func() {
			serde.dbc = nil
		}()
	} else if serde.dbc == nil {
		return nil, errors.New("dbc *DBClient passed into Serialize was null and dbc had not already been set in Serializer state")
	}

	// need to create a new buffer for the serialized record for ptr value calculations,
	// since the incoming buffer (`buf`) already has a lot of stuff written to it (session-id, etc)
	// that are NOT part of the serialized record
	serdebuf := obuf.NewWriteBuffer(80) // holds only the serialized value

	// write the serialization version in so that the buffer size math works
	err := rw.WriteByte(serdebuf, 0)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	err = serde.serializeDocument(serdebuf, doc)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	return serdebuf.Bytes(), nil
}

func getLongFromDB(dbc *DBClient, cmd byte) (int64, error) {
	dbc.buf.Reset()

	err := writeCommandAndSessionId(dbc, cmd)
	if err != nil {
		return int64(-1), oerror.NewTrace(err)
	}

	// send to the OrientDB server
	_, err = dbc.conx.Write(dbc.buf.Bytes())
	if err != nil {
		return int64(-1), oerror.NewTrace(err)
	}

	/* ---[ Read Response ]--- */

	err = readStatusCodeAndSessionId(dbc)
	if err != nil {
		return int64(-1), oerror.NewTrace(err)
	}

	// the answer to the query
	longFromDB, err := rw.ReadLong(dbc.conx)
	if err != nil {
		return int64(-1), oerror.NewTrace(err)
	}

	return longFromDB, nil
}

//
// The link map allow to have as key the types:
// STRING,SHORT,INTEGER,LONG,BYTE,DATE,DECIMAL,DATETIME,DATA,FLOAT,DOUBLE
// the serialization of the linkmap is a list of entry
//
// +----------------------------+
// | values:link_map_entry[]    |
// +----------------------------+
//
// link_map_entry structure
//
// +--------------+------------------+------------+
// | keyType:byte | keyValue:byte[]  | link:LINK  |
// +--------------+------------------+------------+
//
// keyType -  is the type of the key, can be only one of the listed type.
// keyValue - the value of the key serialized with the serializer of the type
// link -     the link value stored with the formant of a LINK
//
// TODO: right now only supporting string keys, but need to support the
//       datatypes listed above (also for EmbeddedMaps)
//
func (serde ORecordSerializerV0) writeLinkMap(buf *obuf.WriteBuf, m map[string]*oschema.OLink) error {
	// number of entries in the map
	err := varint.EncodeAndWriteVarInt32(buf, int32(len(m)))
	if err != nil {
		return oerror.NewTrace(err)
	}

	for k, v := range m {
		// keyType
		err = rw.WriteByte(buf, byte(oschema.STRING))
		if err != nil {
			return oerror.NewTrace(err)
		}
		// keyValue
		err = varint.WriteString(buf, k)
		if err != nil {
			return oerror.NewTrace(err)
		}
		// link
		err = serde.writeLink(buf, v)
		if err != nil {
			return oerror.NewTrace(err)
		}
	}

	return nil
}

func readClassname(buf io.Reader) (string, error) {
	var (
		cnameLen   int32
		cnameBytes []byte
		err        error
	)

	cnameLen, err = varint.ReadVarIntAndDecode32(buf)
	if err != nil {
		return "", oerror.NewTrace(err)
	}
	if cnameLen < 0 {
		return "", oerror.NewTrace(
			fmt.Errorf("Varint for classname len in binary serialization was negative: %d", cnameLen))
	}

	cnameBytes = make([]byte, int(cnameLen))
	n, err := buf.Read(cnameBytes)
	if err != nil {
		return "", oerror.NewTrace(err)
	}
	if n != int(cnameLen) {
		return "",
			fmt.Errorf("Could not read expected number of bytes for className. Expected %d; Read: %d",
				cnameLen, len(cnameBytes))
	}

	return string(cnameBytes), nil
}

//
// readSingleDocument is called by readSingleRecord when it has determined that the server
// has sent a docuemnt ('d'), not flat data ('f') or raw bytes ('b').
// It should be called *after* the single byte below on the first line has been already
// read and determined to be 'd'.  The rest the stream (NOT including the EOT byte) will
// be read.  The serialized document will be turned into an oschema.ODocument.
//
//     Writing byte (1 byte): 100 [OChannelBinaryServer]   <- 'd'=document ('f'=flat data, 'b'=raw bytes)
//     Writing short (2 bytes): 11 [OChannelBinaryServer]  <- cluster-id  (RID part 1)
//     Writing long (8 bytes): 0 [OChannelBinaryServer]    <- cluster-pos (RID part 2)
//     Writing int (4 bytes): 1 [OChannelBinaryServer]     <- version
//     Writing bytes (4+26=30 bytes): [0, 14, 80, 97, 116, ... , 110, 107, 1] <- serialized record
//
func readSingleDocument(dbc *DBClient) (*oschema.ODocument, error) {
	clusterId, err := rw.ReadShort(dbc.conx)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	clusterPos, err := rw.ReadLong(dbc.conx)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	recVersion, err := rw.ReadInt(dbc.conx)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	recBytes, err := rw.ReadBytes(dbc.conx)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}
	rid := oschema.ORID{ClusterID: clusterId, ClusterPos: clusterPos}
	doc, err := createDocumentFromBytes(rid, recVersion, recBytes, dbc)
	ogl.Debugf("::single record doc:::::: %v\n", doc)
	return doc, err
}

//
// Example data section for tree-based LinkBag
//
//                ( --------------------- collectionPointer ----------------------- )  (---size:int--)  (-changes-)
//                (----- fileId:long ----)  ( ---pageIndex:long--- ) (pageOffset:int)
//     TREEBASED             30                         0                 2048                -1             0
//         0,      0, 0, 0, 0, 0, 0, 0, 30,   0, 0, 0, 0, 0, 0, 0, 0,   0, 0, 8, 0,     -1, -1, -1, -1,  0, 0, 0, 0,
//
func readTreeBasedLinkBag(buf io.Reader) (*oschema.OLinkBag, error) {
	fileID, err := rw.ReadLong(buf)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	pageIdx, err := rw.ReadLong(buf)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	pageOffset, err := rw.ReadInt(buf)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	// TODO: need to know how to handle the size and changes stuff => advanced feature not needed yet
	size, err := rw.ReadInt(buf)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	_, err = rw.ReadInt(buf) // changes // TODO: is changes always an int32?
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	return oschema.NewTreeOLinkBag(fileID, pageIdx, pageOffset, size), nil
}

func readEmbeddedLinkBag(rdr io.Reader) (*oschema.OLinkBag, error) {
	bs := make([]byte, 1)
	n, err := rdr.Read(bs)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}
	if n != 1 {
		return nil, oerror.IncorrectNetworkRead{Expected: 1, Actual: n}
	}

	if bs[0] == 1 {
		uuid, err := readLinkBagUUID(rdr)
		if err != nil {
			return nil, oerror.NewTrace(err)
		}
		ogl.Debugf("read uuid %v - now what?\n", uuid)

	} else {
		// if b wasn't zero, then there's no UUID and b was the first byte of an int32
		// specifying the size of the embedded bag collection
		// TODO: I'm not sure this is the right thing - the OrientDB is pretty hazy on how this works
		switch rdr.(type) {
		case *bytes.Buffer:
			buf := rdr.(*bytes.Buffer)
			buf.UnreadByte()

		case *obuf.ReadBuf:
			buf := rdr.(*obuf.ReadBuf)
			buf.UnreadByte()

		default:
			panic("Unknown type of buffer in binserde#readEmbeddedLinkBag")
		}
	}

	bagsz, err := rw.ReadInt(rdr)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}
	links := make([]*oschema.OLink, bagsz)

	for i := int32(0); i < bagsz; i++ {
		clusterID, err := rw.ReadShort(rdr)
		if err != nil {
			return nil, oerror.NewTrace(err)
		}

		clusterPos, err := rw.ReadLong(rdr)
		if err != nil {
			return nil, oerror.NewTrace(err)
		}

		orid := oschema.ORID{ClusterID: clusterID, ClusterPos: clusterPos}
		links[i] = &oschema.OLink{RID: orid}
	}

	return oschema.NewOLinkBag(links), nil
}

//
// readSingleRecord should be called to read a single record from the DBClient connection
// stream (from a db query/command).  In particular, this function should be called
// after the resultType has been read from the stream and resultType == 'r' (byte 114).
// When this is called the 'r' byte shown below should have already been read.  This
// function will then read everything else shown here - including the serialized record,
// but *NOT* including the byte after the serialized record (which is 0 to indicate
// End of Transmission).
//
//     Writing byte (1 byte): 114 [OChannelBinaryServer]   <- 'r' (type=single-record)
//     Writing short (2 bytes): 0 [OChannelBinaryServer]   <- 0=full record  (-2=null, -3=RID only)
//     Writing byte (1 byte): 100 [OChannelBinaryServer]   <- 'd'=document ('f'=flat data, 'b'=raw bytes)
//     Writing short (2 bytes): 11 [OChannelBinaryServer]  <- cluster-id  (RID part 1)
//     Writing long (8 bytes): 0 [OChannelBinaryServer]    <- cluster-pos (RID part 2)
//     Writing int (4 bytes): 1 [OChannelBinaryServer]     <- version
//     Writing bytes (4+26=30 bytes): [0, 14, 80, 97, 116, ... , 110, 107, 1] <- serialized record
//
// A new single ODocument pointer is returned.
//
// TODO: this method needs to determine how to handle 'f' (flat data) and 'b' (raw bytes)
//
func readSingleRecord(dbc *DBClient) (*oschema.ODocument, error) {
	var doc *oschema.ODocument
	resultType, err := rw.ReadShort(dbc.conx)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	if resultType == RecordNull { // null record
		// do nothing - return the zero values of the return types
		return nil, nil

	} else if resultType == RecordRID {
		orid, err := readRID(dbc)
		if err != nil {
			return nil, oerror.NewTrace(err)
		}
		doc = oschema.NewDocument("")
		doc.RID = orid
		ogl.Warn(fmt.Sprintf("readSingleRecord :: Code path not seen before!!: SQLCommand resulted in RID: %s\n", orid))
		// TODO: would now load that record from the DB if the user (Go SQL API) wants it
		return doc, nil

	} else if resultType != int16(0) {
		_, file, line, _ := runtime.Caller(0)
		return nil, fmt.Errorf("Unexpected resultType in SQLCommand (file: %s; line %d): %d",
			file, line+1, resultType)
	}

	// if get here then have a full record, which can be in one of three formats:
	//  - "flat data"
	//  - "raw bytes"
	//  - "document"

	recType, err := rw.ReadByte(dbc.conx)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	if recType == byte('d') {
		return readSingleDocument(dbc)

	} else if recType == byte('f') {
		return readFlatDataRecord(dbc) // ???

	} else if recType == byte('b') {
		return readRawBytesRecord(dbc) // ???

	} else {
		_, file, line, _ := runtime.Caller(0)
		return nil, fmt.Errorf("Unexpected record type. Expected 'd', 'f' or 'b', but was %v (file: %s; line %d)",
			recType, file, line+1)
	}
}

//
// +-----------------+-----------------+
// |clusterId:varint | recordId:varInt |
// +-----------------+-----------------+
//
func (serde ORecordSerializerV0) writeLink(buf *obuf.WriteBuf, lnk *oschema.OLink) error {
	err := varint.EncodeAndWriteVarInt32(buf, int32(lnk.RID.ClusterID))
	if err != nil {
		return oerror.NewTrace(err)
	}

	err = varint.EncodeAndWriteVarInt64(buf, lnk.RID.ClusterPos)
	if err != nil {
		return oerror.NewTrace(err)
	}
	return nil
}

//
// readLink reads a two int64's - the cluster and record.
// We translate it here to a string RID (cluster:record) and return it.
//
func (serde ORecordSerializerV0) readLink(buf io.Reader) (*oschema.OLink, error) {
	clusterID, err := varint.ReadVarIntAndDecode64(buf)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	clusterPos, err := varint.ReadVarIntAndDecode64(buf)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	orid := oschema.ORID{ClusterID: int16(clusterID), ClusterPos: clusterPos}
	return &oschema.OLink{RID: orid}, nil
}

func (ols OLinkSerializer) Deserialize(buf *bytes.Buffer) (interface{}, error) {
	clusterID, err := rw.ReadShort(buf)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	clusterPos, err := rw.ReadLong(buf)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}

	rid := oschema.ORID{ClusterID: clusterID, ClusterPos: clusterPos}
	return &oschema.OLink{RID: rid}, nil
}

//
// readRID should be called when a single record (as opposed to a collection of
// records) is returned from a db query/command (REQUEST_COMMAND only ???).
// That is when the server sends back:
//     1) Writing byte (1 byte): 0 [OChannelBinaryServer]   -> SUCCESS
//     2) Writing int (4 bytes): 192 [OChannelBinaryServer] -> session-id
//     3) Writing byte (1 byte): 114 [OChannelBinaryServer] -> 'r'  (single record)
//     4) Writing short (2 bytes): 0 [OChannelBinaryServer] -> full record (not null, not RID only)
// Line 3 can be 'l' or possibly other things. For 'l' call readResultSet.
// Line 4 can be 0=full-record, -2=null, -3=RID only.  For -3, call readRID.  For 0, call this readSingleDocument.
//
func readRID(dbc *DBClient) (oschema.ORID, error) {
	// svr response: (-3:short)(cluster-id:short)(cluster-position:long)
	// TODO: impl me -> in the future this may need to call loadRecord for the RID and return the ODocument
	clusterID, err := rw.ReadShort(dbc.conx)
	if err != nil {
		return oschema.NewORID(), oerror.NewTrace(err)
	}
	clusterPos, err := rw.ReadLong(dbc.conx)
	if err != nil {
		return oschema.NewORID(), oerror.NewTrace(err)
	}

	return oschema.ORID{ClusterID: clusterID, ClusterPos: clusterPos}, nil
}

func writeLinkBagCollectionPointer(buf *bytes.Buffer, linkBag *oschema.OLinkBag) error {
	// (treePointer:collectionPointer)(changes)
	// where collectionPtr = (fileId:long)(pageIndex:long)(pageOffset:int)
	err := rw.WriteLong(buf, linkBag.GetFileID())
	if err != nil {
		return oerror.NewTrace(err)
	}

	err = rw.WriteLong(buf, linkBag.GetPageIndex())
	if err != nil {
		return oerror.NewTrace(err)
	}

	return rw.WriteInt(buf, linkBag.GetPageOffset())
}

//
// varint.ReadBytes, like rw.ReadBytes, first reads a length from the
// input buffer and then that number of bytes into a []byte from the
// input buffer. The difference is that the integer indicating the length
// of the byte array to follow is a zigzag encoded varint.
//
func ReadBytes(buf io.Reader) ([]byte, error) {
	// an encoded varint give the length of the remaining byte array
	// TODO: might be better to have a ReadVarIntAndDecode that chooses whether to do
	//       int32 or int64 based on the size of the varint and then returns interface{} ?
	lenbytes, err := ReadVarIntAndDecode64(buf)
	if err != nil {
		return nil, err
	}

	if lenbytes == 0 {
		return nil, nil
	}

	if lenbytes < 0 {
		return nil, fmt.Errorf("Error in varint.ReadBytes: size of bytes was less than zero: %v", lenbytes)
	}

	size := int(lenbytes)
	data := make([]byte, size)
	n, err := buf.Read(data)
	if err != nil {
		return nil, oerror.NewTrace(err)
	}
	if n != size {
		return nil, oerror.IncorrectNetworkRead{Expected: size, Actual: n}
	}
	return data, nil
}

//
// varint.ReadString, like rw.ReadString, first reads a length from the
// input buffer and then that number of bytes (of ASCII chars) into a string
// from the input buffer. The difference is that the integer indicating the
// length of the byte array to follow is a zigzag encoded varint.
//
func ReadString(buf io.Reader) (string, error) {
	bs, err := ReadBytes(buf)
	if err != nil {
		return "", oerror.NewTrace(err)
	}
	return string(bs), nil
}

//
// +-------------+-------------------+
// | size:varint | collection:LINK[] |
// +-------------+-------------------+
//
func (serde ORecordSerializerV0) writeLinkList(buf *obuf.WriteBuf, lnks []*oschema.OLink) error {
	// number of entries in the list
	err := varint.EncodeAndWriteVarInt32(buf, int32(len(lnks)))
	if err != nil {
		return oerror.NewTrace(err)
	}

	for _, lnk := range lnks {
		err = serde.writeLink(buf, lnk)
		if err != nil {
			return oerror.NewTrace(err)
		}
	}

	return nil
}

//
// writeDateTime takes an interface{} value that must be of type:
//  - time.Time
//  - int or int64, representing milliseconds since Epoch
//
// NOTE: format for OrientDB DATETIME:
//   Golang formatted date: 2006-01-02 03:04:05
//   Example: 2014-11-25 09:14:54
//
// OrientDB server converts a DATETIME type to millisecond unix epoch and
// stores it as the type LONG.  It is written as a varint long to the
// obuf.WriteBuf passed in.
//
func writeDateTime(buf *obuf.WriteBuf, value interface{}) error {
	var millisEpoch int64

	switch value.(type) {
	case int:
		millisEpoch = int64(value.(int))

	case int64:
		millisEpoch = value.(int64)

	case time.Time:
		// UnixNano returns t as a Unix time, the number of nanoseconds elapsed
		// since January 1, 1970 UTC.
		tm := value.(time.Time)
		tt := tm.Round(time.Millisecond)
		millisEpoch = tt.UnixNano() / (1000 * 1000)

	default:
		return oerror.ErrDataTypeMismatch{
			ExpectedDataType: oschema.DATETIME,
			ExpectedGoType:   "time.Time | int64 | int",
			ActualValue:      value,
		}
	}
	err := varint.EncodeAndWriteVarInt64(buf, millisEpoch)
	if err != nil {
		return oerror.NewTrace(err)
	}
	return nil
}