<commit_before>package gofast

import "sync"
import "sync/atomic"
import "unsafe"
import "fmt"
import "strings"
import "net"
import "time"

import s "github.com/prataprc/gosettings"
import "github.com/prataprc/golog"

type tagfn func(in, out []byte) int
type tagFactory func(*Transport, s.Settings) (uint64, tagfn, tagfn)

var tag_factory = make(map[string]tagFactory)
var transports = unsafe.Pointer(&map[string]*Transporter{})

// RequestCallback handler called for an incoming post, or request,
// or stream message. On either side of the connection, RequestCallback
// initiates a new exchange - be it a Post or Request or Stream type.
// Applications should first register request-handler for expecting
// message-ids, and register a default handler to catch all other
// messages:
//
//   * Stream pointer will be nil if incoming message is a POST.
//   * Handler shall not block and must be as light-weight as possible
//   * If request is initiating a stream of messages from remote,
//   handler should return a stream-callback. StreamCallback will
//   dispatched for every new messages on this stream.
type RequestCallback func(*Stream, BinMessage) StreamCallback

// StreamCallback handler called for an incoming message on a stream,
// the boolean argument, if false, indicates whether remote has closed.
type StreamCallback func(BinMessage, bool)

// Transporter interface to send and receive packets, connection object
// shall implement this interface.
type Transporter interface { // facilitates unit testing
	Read(b []byte) (n int, err error)
	Write(b []byte) (n int, err error)
	LocalAddr() net.Addr
	RemoteAddr() net.Addr
	Close() error
}

// Transport is a peer-to-peer transport enabler.
type Transport struct {
	// statistics, keep this 8-byte aligned.
	n_tx       uint64 // number of packets transmitted
	n_flushes  uint64 // number of times message-batches where flushed
	n_txbyte   uint64 // number of bytes transmitted on socket
	n_txpost   uint64 // number of post messages transmitted
	n_txreq    uint64 // number of request messages transmitted
	n_txresp   uint64 // number of response messages transmitted
	n_txstart  uint64 // number of start messages transmitted
	n_txstream uint64 // number of stream messages transmitted
	n_txfin    uint64 // number of finish messages transmitted
	n_rx       uint64 // number of packets received
	n_rxbyte   uint64 // number of bytes received from socket
	n_rxpost   uint64 // number of post messages received
	n_rxreq    uint64 // number of request messages received
	n_rxresp   uint64 // number of response messages received
	n_rxstart  uint64 // number of start messages received
	n_rxstream uint64 // number of stream messages received
	n_rxfin    uint64 // number of finish messages received
	n_rxbeats  uint64 // number of heartbeats received
	n_dropped  uint64 // number of dropped bytes
	n_mdrops   uint64 // number of dropped messages

	// 0 no handshake
	// 1 oneway handshake
	// 2 bidirectional handshake
	xchngok int64

	// fields.
	name     string
	version  Version
	peerver  atomic.Value
	tagenc   map[uint64]tagfn   // tagid -> func
	tagdec   map[uint64]tagfn   // tagid -> func
	messages map[uint64]Message // msgid -> message
	handlers map[uint64]RequestCallback
	defaulth RequestCallback
	conn     Transporter
	aliveat  int64
	txch     chan *txproto
	rxch     chan rxpacket
	killch   chan struct{}

	// memory pools
	p_strms  chan *Stream // for locally initiated streams
	p_txcmd  chan *txproto
	p_data   chan []byte
	p_rxstrm *sync.Pool

	// settings
	settings   s.Settings
	buffersize uint64
	batchsize  uint64
	chansize   uint64
	logprefix  string
}

//---- transport initialization APIs

// NewTransport encapsulate a transport over this connection,
// one connection one transport.
func NewTransport(
	name string, conn Transporter, version Version,
	setts s.Settings) (*Transport, error) {

	buffersize := setts.Uint64("buffersize")
	opqstart := setts.Uint64("opaque.start")
	opqend := setts.Uint64("opaque.end")
	chansize := setts.Uint64("chansize")
	batchsize := setts.Uint64("batchsize")

	t := &Transport{
		name:    name,
		version: version,
		tagenc:  make(map[uint64]tagfn),
		tagdec:  make(map[uint64]tagfn),
		p_strms: nil, // shall be initialized after setOpaqueRange() call
		p_txcmd: nil, // shall be initialized after setOpaqueRange() call<commit_msg>Fallback to default settings close #17.<commit_after>package gofast

import "sync"
import "sync/atomic"
import "unsafe"
import "fmt"
import "strings"
import "net"
import "time"

import s "github.com/prataprc/gosettings"
import "github.com/prataprc/golog"

type tagfn func(in, out []byte) int
type tagFactory func(*Transport, s.Settings) (uint64, tagfn, tagfn)

var tag_factory = make(map[string]tagFactory)
var transports = unsafe.Pointer(&map[string]*Transporter{})

// RequestCallback handler called for an incoming post, or request,
// or stream message. On either side of the connection, RequestCallback
// initiates a new exchange - be it a Post or Request or Stream type.
// Applications should first register request-handler for expecting
// message-ids, and register a default handler to catch all other
// messages:
//
//   * Stream pointer will be nil if incoming message is a POST.
//   * Handler shall not block and must be as light-weight as possible
//   * If request is initiating a stream of messages from remote,
//   handler should return a stream-callback. StreamCallback will
//   dispatched for every new messages on this stream.
type RequestCallback func(*Stream, BinMessage) StreamCallback

// StreamCallback handler called for an incoming message on a stream,
// the boolean argument, if false, indicates whether remote has closed.
type StreamCallback func(BinMessage, bool)

// Transporter interface to send and receive packets, connection object
// shall implement this interface.
type Transporter interface { // facilitates unit testing
	Read(b []byte) (n int, err error)
	Write(b []byte) (n int, err error)
	LocalAddr() net.Addr
	RemoteAddr() net.Addr
	Close() error
}

// Transport is a peer-to-peer transport enabler.
type Transport struct {
	// statistics, keep this 8-byte aligned.
	n_tx       uint64 // number of packets transmitted
	n_flushes  uint64 // number of times message-batches where flushed
	n_txbyte   uint64 // number of bytes transmitted on socket
	n_txpost   uint64 // number of post messages transmitted
	n_txreq    uint64 // number of request messages transmitted
	n_txresp   uint64 // number of response messages transmitted
	n_txstart  uint64 // number of start messages transmitted
	n_txstream uint64 // number of stream messages transmitted
	n_txfin    uint64 // number of finish messages transmitted
	n_rx       uint64 // number of packets received
	n_rxbyte   uint64 // number of bytes received from socket
	n_rxpost   uint64 // number of post messages received
	n_rxreq    uint64 // number of request messages received
	n_rxresp   uint64 // number of response messages received
	n_rxstart  uint64 // number of start messages received
	n_rxstream uint64 // number of stream messages received
	n_rxfin    uint64 // number of finish messages received
	n_rxbeats  uint64 // number of heartbeats received
	n_dropped  uint64 // number of dropped bytes
	n_mdrops   uint64 // number of dropped messages

	// 0 no handshake
	// 1 oneway handshake
	// 2 bidirectional handshake
	xchngok int64

	// fields.
	name     string
	version  Version
	peerver  atomic.Value
	tagenc   map[uint64]tagfn   // tagid -> func
	tagdec   map[uint64]tagfn   // tagid -> func
	messages map[uint64]Message // msgid -> message
	handlers map[uint64]RequestCallback
	defaulth RequestCallback
	conn     Transporter
	aliveat  int64
	txch     chan *txproto
	rxch     chan rxpacket
	killch   chan struct{}

	// memory pools
	p_strms  chan *Stream // for locally initiated streams
	p_txcmd  chan *txproto
	p_data   chan []byte
	p_rxstrm *sync.Pool

	// settings
	settings   s.Settings
	buffersize uint64
	batchsize  uint64
	chansize   uint64
	logprefix  string
}

//---- transport initialization APIs

// NewTransport encapsulate a transport over this connection,
// one connection one transport.
func NewTransport(
	name string, conn Transporter, version Version,
	setts s.Settings) (*Transport, error) {

	if setts == nil {
		setts = DefaultSettings(1000, 5000)
	}

	buffersize := setts.Uint64("buffersize")
	opqstart := setts.Uint64("opaque.start")
	opqend := setts.Uint64("opaque.end")
	chansize := setts.Uint64("chansize")
	batchsize := setts.Uint64("batchsize")

	t := &Transport{
		name:    name,
		version: version,
		tagenc:  make(map[uint64]tagfn),
		tagdec:  make(map[uint64]tagfn),
		p_strms: nil, // shall be initialized after setOpaqueRange() call
		p_txcmd: nil, // shall be initialized after setOpaqueRange() call
{'fraction_non_alphanumeric': 0.07791027825099375, 'fraction_numerical': 0.018852924474730266, 'mean_word_length': 3.780673181324647, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5470580', 'n_tokens_mistral': 2725, 'n_tokens_neox': 2552, 'n_words': 994}