func (b *bolt4) bufferStream() {
	stream := b.streams.curr
	if stream == nil {
		return
	}
	for {
		rec, batch, _ := b.receiveNext()
		if rec != nil {
			stream.push(rec)
		} else if batch {
			stream.fetchSize = -1
			b.sendPullN()
		} else {
			return
		}
	}
}
Collects all records in current stream if in streaming state and there is a current stream.
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 (
	bolt4_ready        = iota  
	bolt4_streaming            
	bolt4_pendingtx            
	bolt4_tx                   
	bolt4_streamingtx          
	bolt4_failed               
	bolt4_dead                 
	bolt4_unauthorized         
)
const bolt4_fetchsize = 1000
type internalTx4 struct {
	mode         db.AccessMode
	bookmarks    []string
	timeout      time.Duration
	txMeta       map[string]interface{}
	databaseName string
}
func (i *internalTx4) 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
	}
	if i.databaseName != db.DefaultDatabase {
		meta["db"] = i.databaseName
	}
	return meta
}
type bolt4 struct {
	state                 int
	txId                  db.TxHandle
	streams               openstreams
	conn                  net.Conn
	serverName            string
	out                   outgoing
	in                    incoming
	connId                string
	logId                 string
	serverVersion         string
	tfirst                int64        
	pendingTx             internalTx4  
	hasPendingTx          bool
	bookmark              string  
	birthDate             time.Time
	log                   log.Logger
	databaseName          string
	err                   error  
	minor                 int
}
func NewBolt4(serverName string, conn net.Conn, log log.Logger, boltLog log.BoltLogger) *bolt4 {
	b := &bolt4{
		state:      bolt4_unauthorized,
		conn:       conn,
		serverName: serverName,
		birthDate:  time.Now(),
		log:        log,
		streams:    openstreams{},
		in: incoming{
			buf: make([]byte, 4096),
			hyd: hydrator{
				boltLogger: boltLog,
			},
			connReadTimeout: -1,
			logger: log,
		},
	}
	b.out = outgoing{
		chunker:    newChunker(),
		packer:     packstream.Packer{},
		onErr:      func(err error) { b.setError(err, true) },
		boltLogger: boltLog,
	}
	return b
}
func (b *bolt4) checkStreams() {
	if b.streams.num <= 0 {
		switch b.state {
		case bolt4_streamingtx:
			b.state = bolt4_tx
		case bolt4_streaming:
			b.state = bolt4_ready
		}
	}
}
func (b *bolt4) ServerName() string {
	return b.serverName
}
func (b *bolt4) ServerVersion() string {
	return b.serverVersion
}
func (b *bolt4) setError(err error, fatal bool) {
	if err == nil {
		return
	}
	if b.err == nil {
		b.err = err
		b.state = bolt4_failed
	}
	neo4jErr, _ := err.(*db.Neo4jError)
	if fatal || (neo4jErr != nil && neo4jErr.Code == "Status.Security.AuthorizationExpired") {
		b.state = bolt4_dead
	}
	if b.streams.curr != nil {
		b.streams.detach(nil, err)
		b.checkStreams()
	}
	if neo4jErr != nil && neo4jErr.Classification() == "ClientError" {
		b.log.Debugf(log.Bolt4, b.logId, "%s", err)
	} else {
		b.log.Error(log.Bolt4, b.logId, err)
	}
}
func (b *bolt4) receiveMsg() interface{} {
	if b.err != nil {
		return nil
	}
	msg, err := b.in.next(b.conn)
	b.setError(err, true)
	return msg
}
func (b *bolt4) receiveSuccess() *success {
	msg := b.receiveMsg()
	if b.err != nil {
		return nil
	}
	switch v := msg.(type) {
	case *success:
		return v
	case *db.Neo4jError:
		b.setError(v, false)
		return nil
	default:
		b.setError(errors.New("Expected success or database error"), true)
		return nil
	}
}
func (b *bolt4) connect(minor int, auth map[string]interface{}, userAgent string, routingContext map[string]string) error {
	if err := b.assertState(bolt4_unauthorized); err != nil {
		return err
	}
	hello := map[string]interface{}{
		"user_agent": userAgent,
	}
	if minor >= 1 {
		if routingContext != nil {
			hello["routing"] = routingContext
		}
	}
	for k, v := range auth {
		_, exists := hello[k]
		if !exists {
			hello[k] = v
		}
	}
	b.out.appendHello(hello)
	b.out.send(b.conn)
	succ := b.receiveSuccess()
	if b.err != nil {
		return b.err
	}
	b.connId = succ.connectionId
	b.serverVersion = succ.server
	connectionLogId := fmt.Sprintf("%s@%s", b.connId, b.serverName)
	b.logId = connectionLogId
	b.in.hyd.logId = connectionLogId
	b.in.logId = connectionLogId
	b.out.logId = connectionLogId
	b.initializeReadTimeoutHint(succ.configurationHints)
	b.state = bolt4_ready
	b.minor = minor
	b.streams.reset()
	b.log.Infof(log.Bolt4, b.logId, "Connected")
	return nil
}
func (b *bolt4) TxBegin(txConfig db.TxConfig) (db.TxHandle, error) {
	if b.state == bolt4_streaming {
		if b.bufferStream(); b.err != nil {
			return 0, b.err
		}
	}
	b.streams.reset()
	if err := b.assertState(bolt4_ready); err != nil {
		return 0, err
	}
	tx := internalTx4{
		mode:         txConfig.Mode,
		bookmarks:    txConfig.Bookmarks,
		timeout:      txConfig.Timeout,
		txMeta:       txConfig.Meta,
		databaseName: b.databaseName,
	}
	if len(tx.bookmarks) > 0 {
		b.out.appendBegin(tx.toMeta())
		b.out.send(b.conn)
		b.receiveSuccess()
		if b.err != nil {
			return 0, b.err
		}
		b.state = bolt4_tx
		b.hasPendingTx = false
	} else {
		b.pendingTx = tx
		b.hasPendingTx = true
		b.state = bolt4_pendingtx
	}
	b.txId = db.TxHandle(time.Now().Unix())
	return b.txId, nil
}
func (b *bolt4) assertTxHandle(h1, h2 db.TxHandle) error {
	if h1 != h2 {
		err := errors.New("Invalid transaction handle")
		b.log.Error(log.Bolt4, b.logId, err)
		return err
	}
	return nil
}
func (b *bolt4) 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.Bolt4, b.logId, err)
	return err
}
func (b *bolt4) TxCommit(txh db.TxHandle) error {
	if err := b.assertTxHandle(b.txId, txh); err != nil {
		return err
	}
	if b.state == bolt4_pendingtx {
		b.state = bolt4_ready
		return nil
	}
	if b.discardAllStreams(); b.err != nil {
		return b.err
	}
	if err := b.assertState(bolt4_tx); err != nil {
		return err
	}
	b.out.appendCommit()
	b.out.send(b.conn)
	succ := b.receiveSuccess()
	if b.err != nil {
		return b.err
	}
	if len(succ.bookmark) > 0 {
		b.bookmark = succ.bookmark
	}
	b.state = bolt4_ready
	return nil
}
func (b *bolt4) TxRollback(txh db.TxHandle) error {
	if err := b.assertTxHandle(b.txId, txh); err != nil {
		return err
	}
	if b.state == bolt4_pendingtx {
		b.state = bolt4_ready
		return nil
	}
	if b.discardAllStreams(); b.err != nil {
		return b.err
	}
	if err := b.assertState(bolt4_tx); err != nil {
		return err
	}
	b.out.appendRollback()
	b.out.send(b.conn)
	if b.receiveSuccess(); b.err != nil {
		return b.err
	}
	b.state = bolt4_ready
	return nil
}
func (b *bolt4) discardStream() {
	if b.state != bolt4_streaming && b.state != bolt4_streamingtx {
		return
	}
	stream := b.streams.curr
	if stream == nil {
		return
	}
	discarded := false
	for {
		_, batch, sum := b.receiveNext()
		if batch {
			if discarded {
				b.streams.remove(stream)
				b.checkStreams()
				return
			}
			discarded = true
			stream.fetchSize = -1
			if b.state == bolt4_streamingtx {
				b.out.appendDiscardNQid(stream.fetchSize, stream.qid)
			} else {
				b.out.appendDiscardN(stream.fetchSize)
			}
			b.out.send(b.conn)
		} else if sum != nil || b.err != nil {
			return
		}
	}
}
func (b *bolt4) discardAllStreams() {
	if b.state != bolt4_streaming && b.state != bolt4_streamingtx {
		return
	}
	b.discardStream()
	b.streams.reset()
	b.checkStreams()
}
func (b *bolt4) sendPullN() {
	b.assertState(bolt4_streaming, bolt3_streamingtx)
	if b.state == bolt4_streaming {
		b.out.appendPullN(b.streams.curr.fetchSize)
		b.out.send(b.conn)
	} else if b.state == bolt4_streamingtx {
		b.out.appendPullNQid(b.streams.curr.fetchSize, b.streams.curr.qid)
		b.out.send(b.conn)
	}
}