<commit_before>package rpc

import (
	"io"

	"github.com/ugorji/go/codec"
)

type decoder interface {
	Decode(interface{}) error
}

// TODO rename this interface to EncodeAndSend
type encoder interface {
	Encode(interface{}) <-chan error
}

type decoderWrapper struct {
	*codec.Decoder
	fieldNumber int
}

func newDecoderWrapper() *decoderWrapper {
	return &decoderWrapper{
		Decoder:     codec.NewDecoderBytes([]byte{}, newCodecMsgpackHandle()),
		fieldNumber: 0,
	}
}

func (dw *decoderWrapper) Decode(i interface{}) error {
	defer func() {
		dw.fieldNumber++
	}()

	err := dw.Decoder.Decode(i)
	if err != nil {
		return newRPCMessageFieldDecodeError(dw.fieldNumber, err)
	}
	return nil
}

func newCodecMsgpackHandle() codec.Handle {
	return &codec.MsgpackHandle{
		WriteExt:    true,
		RawToString: true,
	}
}

type writeBundle struct {
	bytes []byte
	ch    chan error
}

type framedMsgpackEncoder struct {
	handle   codec.Handle
	writer   io.Writer
	writeCh  chan writeBundle
	doneCh   chan struct{}
	closedCh chan struct{}
}

func newFramedMsgpackEncoder(writer io.Writer) *framedMsgpackEncoder {
	e := &framedMsgpackEncoder{
		handle:   newCodecMsgpackHandle(),
		writer:   writer,
		writeCh:  make(chan writeBundle),
		doneCh:   make(chan struct{}),
		closedCh: make(chan struct{}),
	}
	go e.writerLoop()
	return e
}

func (e *framedMsgpackEncoder) encodeToBytes(enc *codec.Encoder, i interface{}) (v []byte, err error) {
	enc.ResetBytes(&v)
	err = enc.Encode(i)
	return v, err
}

func (e *framedMsgpackEncoder) encodeFrame(i interface{}) ([]byte, error) {
	enc := codec.NewEncoderBytes(&[]byte{}, e.handle)
	content, err := e.encodeToBytes(enc, i)
	if err != nil {
		return nil, err
	}
	length, err := e.encodeToBytes(enc, len(content))
	if err != nil {
		return nil, err<commit_msg>Make the write channel buffered to prevent starvation<commit_after>package rpc

import (
	"io"

	"github.com/ugorji/go/codec"
)

type decoder interface {
	Decode(interface{}) error
}

// TODO rename this interface to EncodeAndSend
type encoder interface {
	Encode(interface{}) <-chan error
}

type decoderWrapper struct {
	*codec.Decoder
	fieldNumber int
}

func newDecoderWrapper() *decoderWrapper {
	return &decoderWrapper{
		Decoder:     codec.NewDecoderBytes([]byte{}, newCodecMsgpackHandle()),
		fieldNumber: 0,
	}
}

func (dw *decoderWrapper) Decode(i interface{}) error {
	defer func() {
		dw.fieldNumber++
	}()

	err := dw.Decoder.Decode(i)
	if err != nil {
		return newRPCMessageFieldDecodeError(dw.fieldNumber, err)
	}
	return nil
}

func newCodecMsgpackHandle() codec.Handle {
	return &codec.MsgpackHandle{
		WriteExt:    true,
		RawToString: true,
	}
}

type writeBundle struct {
	bytes []byte
	ch    chan error
}

type framedMsgpackEncoder struct {
	handle   codec.Handle
	writer   io.Writer
	writeCh  chan writeBundle
	doneCh   chan struct{}
	closedCh chan struct{}
}

func newFramedMsgpackEncoder(writer io.Writer) *framedMsgpackEncoder {
	e := &framedMsgpackEncoder{
		handle:   newCodecMsgpackHandle(),
		writer:   writer,
		writeCh:  make(chan writeBundle, 128),
		doneCh:   make(chan struct{}),
		closedCh: make(chan struct{}),
	}
	go e.writerLoop()
	return e
}

func (e *framedMsgpackEncoder) encodeToBytes(enc *codec.Encoder, i interface{}) (v []byte, err error) {
	enc.ResetBytes(&v)
	err = enc.Encode(i)
	return v, err
}

func (e *framedMsgpackEncoder) encodeFrame(i interface{}) ([]byte, error) {
	enc := codec.NewEncoderBytes(&[]byte{}, e.handle)
	content, err := e.encodeToBytes(enc, i)
	if err != nil {
		return nil, err
	}
	length, err := e.encodeToBytes(enc, len(content))
	if err != nil {
		return nil, err
{'fraction_non_alphanumeric': 0.12379376895505928, 'fraction_numerical': 0.0013785497656465398, 'mean_word_length': 4.455639097744361, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, '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': '6728287', 'n_tokens_mistral': 1298, 'n_tokens_neox': 1231, 'n_words': 327}