<commit_before>package flux

import (
	"bytes"
	"errors"
	"io"
)

type (

	//StreamMod is a function that modifies a data []byte
	StreamMod func(bu []byte) []byte

	//StreamInterface define the interface method rules for streams
	StreamInterface interface {
		Subscribe(func(interface{}, *Sub)) *Sub
		StreamMod(fx func(interface{}) interface{}) (StreamInterface, error)
		Stream() (StreamInterface, error)
		StreamWriter(io.Writer) *Sub
		StreamReader(io.Reader) *Sub
		Write([]byte) (int, error)
		Read([]byte) (int, error)
		Emit(interface{}) (int, error)
		Close() error
		OnClosed() ActionInterface
	}

	//BaseStream defines a basic stream structure
	BaseStream struct {
		push   Pipe
		closed ActionInterface
		sub    *Sub
	}

	//ByteStream provides a buffer back streamer
	ByteStream struct {
		StreamInterface
		buf *bytes.Buffer
	}

	//WrapByteStream provides a ByteStream wrapped around a reader
	WrapByteStream struct {
		StreamInterface
		reader io.ReadCloser
	}
)

var (
	//ErrWrongType denotes a wrong-type assertion
	ErrWrongType = errors.New("Wrong Type!")
)

//NewBaseStream returns a basestream instance
func NewBaseStream() *BaseStream {
	return &BaseStream{PushSocket(0), NewAction(), nil}
}

//OnClosed returns an action that gets fullfilled when the stream is closed
func (b *BaseStream) OnClosed() ActionInterface {
	return b.closed.Wrap()
}

//Close closes the stream
func (b *BaseStream) Close() error {
	if b.sub != nil {
		b.sub.Close()
	}
	b.push.Close()
	b.closed.Fullfill(true)
	return nil
}

//Stream creates a new StreamInterface and pipes all current data into that stream
func (b *BaseStream) Stream() (StreamInterface, error) {
	return &BaseStream{PushSocketWith(b.push), NewAction(), nil}, nil
}

//StreamMod provides a new stream that mods the internal details of these streams byte
func (b *BaseStream) StreamMod(fn func(v interface{}) interface{}) (StreamInterface, error) {
	return &BaseStream{DoPushSocket(b.push, func(v interface{}, sock SocketInterface) {<commit_msg>added Pull Socket<commit_after>package flux

import (
	"bytes"
	"errors"
	"io"
)

type (

	//StreamMod is a function that modifies a data []byte
	StreamMod func(bu []byte) []byte

	//StreamInterface define the interface method rules for streams
	StreamInterface interface {
		Subscribe(func(interface{}, *Sub)) *Sub
		StreamMod(fx func(interface{}) interface{}) (StreamInterface, error)
		Stream() (StreamInterface, error)
		StreamWriter(io.Writer) *Sub
		StreamReader(io.Reader) *Sub
		Write([]byte) (int, error)
		Read([]byte) (int, error)
		Emit(interface{}) (int, error)
		Close() error
		OnClosed() ActionInterface
	}

	//BaseStream defines a basic stream structure
	BaseStream struct {
		push   Pipe
		closed ActionInterface
		sub    *Sub
	}

	//ByteStream provides a buffer back streamer
	ByteStream struct {
		StreamInterface
		buf *bytes.Buffer
	}

	//WrapByteStream provides a ByteStream wrapped around a reader
	WrapByteStream struct {
		StreamInterface
		reader io.ReadCloser
	}
)

var (
	//ErrWrongType denotes a wrong-type assertion
	ErrWrongType = errors.New("Wrong Type!")
	//ErrReadMisMatch denotes when the total bytes length is not read
	ErrReadMisMatch = errors.New("Length MisMatch,Data not fully Read")
)

//WrapByteStreamWith returns a bytestream that wraps a reader closer
func WrapByteStreamWith(rd io.ReadCloser) *WrapByteStream {
	return &WrapByteStream{NewByteStream(), rd}
}

//Read reads the data from internal wrap ReadCloser.
func (b *WrapByteStream) Read(data []byte) (int, error) {
	var nx int
	var errx error
	var err error

	nx, errx = b.reader.Read(data)

	if errx != nil {
		nx, err = b.StreamInterface.Read(data)

		if err == nil {
			errx = nil
		}
	}

	return nx, errx
}

//NewBaseStream returns a basestream instance
func NewBaseStream() *BaseStream {
	return &BaseStream{PushSocket(0), NewAction(), nil}
}

//OnClosed returns an action that gets fullfilled when the stream is closed
func (b *BaseStream) OnClosed() ActionInterface {
	return b.closed.Wrap()
}

//Close closes the stream
func (b *BaseStream) Close() error {
	if b.sub != nil {
		b.sub.Close()
	}
	b.push.Close()
	b.closed.Fullfill(true)
	return nil
}

//Stream creates a new StreamInterface and pipes all current data into that stream
func (b *BaseStream) Stream() (StreamInterface, error) {
	return &BaseStream{PushSocketWith(b.push), NewAction(), nil}, nil
}

//StreamMod provides a new stream that mods the internal details of these streams byte
func (b *BaseStream) StreamMod(fn func(v interface{}) interface{}) (StreamInterface, error) {
	return &BaseStream{DoPushSocket(b.push, func(v interface{}, sock SocketInterface) {
{'fraction_non_alphanumeric': 0.10721205597416578, 'fraction_numerical': 0.0004305705059203445, 'mean_word_length': 4.557416267942584, 'pattern_counts': {'":': 0, '<': 3, '<?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': '4060340', 'n_tokens_mistral': 1497, 'n_tokens_neox': 1427, 'n_words': 487}