func (b *bolt3) Close() {
	b.log.Infof(log.Bolt3, b.logId, "Close")
	if b.state != bolt3_dead {
		b.out.appendGoodbye()
		b.out.send(b.conn)
	}
	b.conn.Close()
	b.state = bolt3_dead
}
Beware, could be called on another thread when driver is closed.
package bolt
import (
	"errors"
	"fmt"
	"net"
	"time"
	"github.com/neo4j/neo4j-go-driver/v4/neo4j/db"
	"github.com/neo4j/neo4j-go-driver/v4/neo4j/internal/packstream"
	"github.com/neo4j/neo4j-go-driver/v4/neo4j/log"
)
const (
	bolt3_ready        = iota  
	bolt3_streaming            
	bolt3_pendingtx            
	bolt3_tx                   
	bolt3_streamingtx          
	bolt3_failed               
	bolt3_dead                 
	bolt3_unauthorized         
)
type internalTx3 struct {
	mode      db.AccessMode
	bookmarks []string
	timeout   time.Duration
	txMeta    map[string]interface{}
}
func (i *internalTx3) toMeta() map[string]interface{} {
	meta := map[string]interface{}{}
	if i.mode == db.ReadMode {
		meta["mode"] = "r"
	}
	if len(i.bookmarks) > 0 {
		meta["bookmarks"] = i.bookmarks
	}
	ms := int(i.timeout.Nanoseconds() / 1e6)
	if ms > 0 {
		meta["tx_timeout"] = ms
	}
	if len(i.txMeta) > 0 {
		meta["tx_metadata"] = i.txMeta
	}
	return meta
}
type bolt3 struct {
	state         int
	txId          db.TxHandle
	currStream    *stream
	conn          net.Conn
	serverName    string
	out           *outgoing
	in            *incoming
	connId        string
	logId         string
	serverVersion string
	tfirst        int64         
	pendingTx     *internalTx3  
	bookmark      string        
	birthDate     time.Time
	log           log.Logger
	err           error  
	minor         int
}
func NewBolt3(serverName string, conn net.Conn, log log.Logger, boltLog log.BoltLogger) *bolt3 {
	b := &bolt3{
		state:      bolt3_unauthorized,
		conn:       conn,
		serverName: serverName,
		in: &incoming{
			buf: make([]byte, 4096),
			hyd: hydrator{
				boltLogger: boltLog,
			},
			connReadTimeout: -1,
			logger: log,
		},
		birthDate: time.Now(),
		log:       log,
	}
	b.out = &outgoing{
		chunker: newChunker(),
		packer:  packstream.Packer{},
		onErr: func(err error) {
			if b.err == nil {
				b.err = err
			}
			b.state = bolt3_dead
		},
		boltLogger: boltLog,
	}
	return b
}
func (b *bolt3) ServerName() string {
	return b.serverName
}
func (b *bolt3) ServerVersion() string {
	return b.serverVersion
}
func (b *bolt3) receiveMsg() interface{} {
	msg, err := b.in.next(b.conn)
	if err != nil {
		b.err = err
		b.log.Error(log.Bolt3, b.logId, b.err)
		b.state = bolt3_dead
		return nil
	}
	return msg
}
func (b *bolt3) receiveSuccess() *success {
	switch v := b.receiveMsg().(type) {
	case *success:
		return v
	case *db.Neo4jError:
		b.state = bolt3_failed
		b.err = v
		if v.Classification() == "ClientError" {
			b.log.Debugf(log.Bolt3, b.logId, "%s", v)
		} else {
			b.log.Error(log.Bolt3, b.logId, v)
		}
		return nil
	default:
		if b.err != nil {
			return nil
		}
		b.state = bolt3_dead
		b.err = errors.New("Expected success or database error")
		b.log.Error(log.Bolt3, b.logId, b.err)
		return nil
	}
}
func (b *bolt3) connect(minor int, auth map[string]interface{}, userAgent string) error {
	if err := b.assertState(bolt3_unauthorized); err != nil {
		return err
	}
	hello := map[string]interface{}{
		"user_agent": userAgent,
	}
	for k, v := range auth {
		_, exists := hello[k]
		if exists {
			continue
		}
		hello[k] = v
	}
	b.out.appendHello(hello)
	if b.out.send(b.conn); b.err != nil {
		return b.err
	}
	succ := b.receiveSuccess()
	if b.err != nil {
		return b.err
	}
	b.connId = succ.connectionId
	connectionLogId := fmt.Sprintf("%s@%s", b.connId, b.serverName)
	b.logId = connectionLogId
	b.in.logId = connectionLogId
	b.in.hyd.logId = connectionLogId
	b.out.logId = connectionLogId
	b.serverVersion = succ.server
	b.state = bolt3_ready
	b.minor = minor
	b.log.Infof(log.Bolt3, b.logId, "Connected")
	return nil
}
func (b *bolt3) TxBegin(txConfig db.TxConfig) (db.TxHandle, error) {
	if b.state == bolt3_streaming {
		if err := b.bufferStream(); err != nil {
			return 0, err
		}
	}
	if err := b.assertState(bolt3_ready); err != nil {
		return 0, err
	}
	tx := &internalTx3{
		mode:      txConfig.Mode,
		bookmarks: txConfig.Bookmarks,
		timeout:   txConfig.Timeout,
		txMeta:    txConfig.Meta,
	}
	if len(tx.bookmarks) > 0 {
		b.out.appendBegin(tx.toMeta())
		if b.out.send(b.conn); b.err != nil {
			return 0, b.err
		}
		if b.receiveSuccess(); b.err != nil {
			return 0, b.err
		}
		b.state = bolt3_tx
	} else {
		b.pendingTx = tx
		b.state = bolt3_pendingtx
	}
	b.txId = db.TxHandle(time.Now().Unix())
	return b.txId, nil
}
func (b *bolt3) assertTxHandle(h1, h2 db.TxHandle) error {
	if h1 != h2 {
		err := errors.New("Invalid transaction handle")
		b.log.Error(log.Bolt3, b.logId, err)
		return err
	}
	return nil
}
func (b *bolt3) assertState(allowed ...int) error {
	if b.err != nil {
		return b.err
	}
	for _, a := range allowed {
		if b.state == a {
			return nil
		}
	}
	err := errors.New(fmt.Sprintf("Invalid state %d, expected: %+v", b.state, allowed))
	b.log.Error(log.Bolt3, b.logId, err)
	return err
}
func (b *bolt3) TxCommit(txh db.TxHandle) error {
	if err := b.assertTxHandle(b.txId, txh); err != nil {
		return err
	}
	if b.state == bolt3_pendingtx {
		b.state = bolt3_ready
		return nil
	}
	if b.state == bolt3_streamingtx {
		if err := b.discardStream(); err != nil {
			return err
		}
	}
	if err := b.assertState(bolt3_tx); err != nil {
		return err
	}
	b.out.appendCommit()
	if b.out.send(b.conn); b.err != nil {
		return b.err
	}
	succ := b.receiveSuccess()
	if b.err != nil {
		return b.err
	}
	if len(succ.bookmark) > 0 {
		b.bookmark = succ.bookmark
	}
	b.state = bolt3_ready
	return nil
}
func (b *bolt3) TxRollback(txh db.TxHandle) error {
	if err := b.assertTxHandle(b.txId, txh); err != nil {
		return err
	}
	if b.state == bolt3_pendingtx {
		b.state = bolt3_ready
		return nil
	}
	if b.state == bolt3_streamingtx {
		if err := b.discardStream(); err != nil {
			return err
		}
	}
	if err := b.assertState(bolt3_tx); err != nil {
		return err
	}
	b.out.appendRollback()
	if b.out.send(b.conn); b.err != nil {
		return b.err
	}
	if b.receiveSuccess(); b.err != nil {
		return b.err
	}
	b.state = bolt3_ready
	return nil
}
func (b *bolt3) discardStream() error {
	if b.state != bolt3_streaming && b.state != bolt3_streamingtx {
		return nil
	}
	var (
		sum *db.Summary
		err error
	)
	for sum == nil && err == nil {
		_, sum, err = b.receiveNext()
	}
	return err
}
func (b *bolt3) bufferStream() error {
	if b.state != bolt3_streaming && b.state != bolt3_streamingtx {
		return nil
	}
	n := 0
	var (
		sum *db.Summary
		err error
		rec *db.Record
	)
	for sum == nil && err == nil {
		rec, sum, err = b.receiveNext()
		if rec != nil {
			b.currStream.push(rec)
			n++
		}
	}
	if n > 0 {
		b.log.Warnf(log.Bolt3, b.logId, "Buffered %d records", n)
	}
	return err
}
func (b *bolt3) run(cypher string, params map[string]interface{}, tx *internalTx3) (*stream, error) {
	if err := b.bufferStream(); err != nil {
		return nil, err
	}
	if err := b.assertState(bolt3_tx, bolt3_ready, bolt3_pendingtx); err != nil {
		return nil, err
	}
	var meta map[string]interface{}
	if tx != nil {
		meta = tx.toMeta()
	}
	if b.state == bolt3_pendingtx {
		b.out.appendBegin(meta)
		meta = nil
	}
	b.out.appendRun(cypher, params, meta)
	b.out.appendPullAll()
	if b.out.send(b.conn); b.err != nil {
		return nil, b.err
	}
	if b.state == bolt3_pendingtx {
		if b.receiveSuccess(); b.err != nil {
			return nil, b.err
		}
		b.state = bolt3_tx
	}
	succ := b.receiveSuccess()
	if b.err != nil {
		return nil, b.err
	}
	b.tfirst = succ.tfirst
	if b.state == bolt3_ready {
		b.state = bolt3_streaming
	} else {
		b.state = bolt3_streamingtx
	}
	b.currStream = &stream{keys: succ.fields}
	return b.currStream, nil
}
func (b *bolt3) Run(runCommand db.Command, txConfig db.TxConfig) (db.StreamHandle, error) {
	if err := b.assertState(bolt3_streaming, bolt3_ready); err != nil {
		return nil, err
	}
	tx := internalTx3{
		mode:      txConfig.Mode,
		bookmarks: txConfig.Bookmarks,
		timeout:   txConfig.Timeout,
		txMeta:    txConfig.Meta,
	}
	stream, err := b.run(runCommand.Cypher, runCommand.Params, &tx)
	if err != nil {
		return nil, err
	}
	return stream, nil
}
func (b *bolt3) RunTx(txh db.TxHandle, runCommand db.Command) (db.StreamHandle, error) {
	if err := b.assertTxHandle(b.txId, txh); err != nil {
		return nil, err
	}
	stream, err := b.run(runCommand.Cypher, runCommand.Params, b.pendingTx)
	b.pendingTx = nil
	if err != nil {
		return nil, err
	}
	return stream, nil
}
func (b *bolt3) Keys(streamHandle db.StreamHandle) ([]string, error) {
	stream, ok := streamHandle.(*stream)
	if !ok {
		return nil, errors.New("Invalid stream handle")
	}
	return stream.keys, nil
}
func (b *bolt3) Next(streamHandle db.StreamHandle) (*db.Record, *db.Summary, error) {
	stream, ok := streamHandle.(*stream)
	if !ok {
		return nil, nil, errors.New("Invalid stream handle")
	}
	buf, rec, sum, err := stream.bufferedNext()
	if buf {
		return rec, sum, err
	}
	if stream != b.currStream {
		return nil, nil, errors.New("Invalid stream handle")
	}
	return b.receiveNext()
}
func (b *bolt3) Consume(streamHandle db.StreamHandle) (*db.Summary, error) {
	stream, ok := streamHandle.(*stream)
	if !ok {
		return nil, errors.New("Invalid stream handle")
	}
	if stream != b.currStream {
		return stream.sum, stream.err
	}
	if stream.err != nil || stream.sum != nil {
		return stream.sum, stream.err
	}
	b.discardStream()
	return stream.sum, stream.err
}
func (b *bolt3) Buffer(streamHandle db.StreamHandle) error {
	stream, ok := streamHandle.(*stream)
	if !ok {
		return errors.New("Invalid stream handle")
	}
	if stream != b.currStream {
		return stream.Err()
	}
	if stream.err != nil || stream.sum != nil {
		return stream.Err()
	}
	b.bufferStream()
	return stream.Err()
}
func (b *bolt3) receiveNext() (*db.Record, *db.Summary, error) {
	if err := b.assertState(bolt3_streaming, bolt3_streamingtx); err != nil {
		return nil, nil, err
	}
	res := b.receiveMsg()
	if b.err != nil {
		return nil, nil, b.err
	}
	switch x := res.(type) {
	case *db.Record:
		x.Keys = b.currStream.keys
		return x, nil, nil
	case *success:
		sum := x.summary()
		if sum == nil {
			b.state = bolt3_dead
			b.err = errors.New("Failed to parse summary")
			b.currStream.err = b.err
			b.currStream = nil
			b.log.Error(log.Bolt3, b.logId, b.err)
			return nil, nil, b.err
		}
		if b.state == bolt3_streamingtx {
			b.state = bolt3_tx
		} else {
			b.state = bolt3_ready
			if len(sum.Bookmark) > 0 {
				b.bookmark = sum.Bookmark
			}
		}
		b.currStream.sum = sum
		b.currStream = nil
		sum.Agent = b.serverVersion
		sum.Major = 3
		sum.Minor = b.minor
		sum.ServerName = b.serverName
		sum.TFirst = b.tfirst
		return nil, sum, nil
	case *db.Neo4jError:
		b.err = x
		b.currStream.err = b.err
		b.currStream = nil
		b.state = bolt3_failed
		if x.Classification() == "ClientError" {
			b.log.Debugf(log.Bolt3, b.logId, "%s", x)
		} else {
			b.log.Error(log.Bolt3, b.logId, x)
		}
		return nil, nil, x
	default:
		b.state = bolt3_dead
		b.err = errors.New("Unknown response")
		b.currStream.err = b.err
		b.currStream = nil
		b.log.Error(log.Bolt3, b.logId, b.err)
		return nil, nil, b.err
	}
}
func (b *bolt3) Bookmark() string {
	return b.bookmark
}
func (b *bolt3) IsAlive() bool {
	return b.state != bolt3_dead
}
func (b *bolt3) Birthdate() time.Time {
	return b.birthDate
}
func (b *bolt3) Reset() {
	defer func() {
		b.txId = 0
		b.currStream = nil
		b.bookmark = ""
		b.pendingTx = nil
		b.err = nil
	}()
	if b.state == bolt3_ready || b.state == bolt3_dead {
		return
	}
	b.discardStream()
	if b.state == bolt3_ready || b.state == bolt3_dead {
		return
	}
	b.err = nil
	b.out.appendReset()
	if b.out.send(b.conn); b.err != nil {
		return
	}
	for {
		msg := b.receiveMsg()
		if b.err != nil {
			return
		}
		switch msg.(type) {
		case *ignored:
		case *success:
			b.state = bolt3_ready
			return
		default:
			b.state = bolt3_dead
			return
		}
	}
}
func (b *bolt3) GetRoutingTable(context map[string]string, bookmarks []string, database string) (*db.RoutingTable, error) {
	if err := b.assertState(bolt3_ready); err != nil {
		return nil, err
	}
	if database != db.DefaultDatabase {
		return nil, errors.New("Bolt 3 does not support routing to a specifiec database name")
	}
	runCommand := db.Command{
		Cypher: "CALL dbms.cluster.routing.getRoutingTable($context)",
		Params: map[string]interface{}{"context": context},
	}
	txConfig := db.TxConfig{Mode: db.ReadMode}
	streamHandle, err := b.Run(runCommand, txConfig)
	if err != nil {
		dbError, isDbError := err.(*db.Neo4jError)
		if isDbError && dbError.Code == "Neo.ClientError.Procedure.ProcedureNotFound" {
			return nil, &db.RoutingNotSupportedError{Server: b.serverName}
		}
		return nil, err
	}
	rec, _, err := b.Next(streamHandle)
	if err != nil {
		return nil, err
	}
	if rec == nil {
		return nil, errors.New("No routing table record")
	}
	b.Next(streamHandle)
	table := parseRoutingTableRecord(rec)
	if table == nil {
		return nil, errors.New("Unable to parse routing table")
	}
	return table, nil
}