142 lines
3.3 KiB
Go
142 lines
3.3 KiB
Go
package packer
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"io"
|
|
"sync"
|
|
)
|
|
|
|
// TODO: try to be more like encoding/csv, or more like encoding/pem and encoding/json?
|
|
|
|
// Encoder converts TCP to MPLEXY-TCP
|
|
type Encoder struct {
|
|
ctx context.Context
|
|
subctx context.Context
|
|
mux sync.Mutex
|
|
//out io.WriteCloser
|
|
out io.Writer
|
|
outErr chan error
|
|
bufferSize int
|
|
}
|
|
|
|
// NewEncoder returns an Encoder instance
|
|
func NewEncoder(ctx context.Context, wout io.Writer) *Encoder {
|
|
enc := &Encoder{
|
|
ctx: ctx,
|
|
out: wout,
|
|
outErr: make(chan error),
|
|
bufferSize: defaultBufferSize,
|
|
}
|
|
return enc
|
|
}
|
|
|
|
// Run loops over a select of contexts and error channels
|
|
// to cancel and close south-side connections, if needed.
|
|
// TODO should this be pushed elsewhere to handled?
|
|
func (enc *Encoder) Run() error {
|
|
ctx, cancel := context.WithCancel(enc.ctx)
|
|
defer cancel()
|
|
|
|
enc.subctx = ctx
|
|
|
|
for {
|
|
select {
|
|
// TODO: do children respond to children cancelling?
|
|
case <-enc.ctx.Done():
|
|
// TODO
|
|
//_ = enc.out.Close()
|
|
return errors.New("context cancelled")
|
|
case err := <-enc.outErr:
|
|
// if a write fails for one, it fail for all
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
|
|
// Encode adds MPLEXY headers to raw net traffic, and is intended to be used on each client connection
|
|
func (enc *Encoder) Encode(rin io.Reader, src, dst Addr) error {
|
|
rx := make(chan []byte)
|
|
rxErr := make(chan error)
|
|
|
|
//fmt.Println("[debug] what's the source to encode?", src)
|
|
|
|
go func() {
|
|
for {
|
|
b := make([]byte, enc.bufferSize)
|
|
//fmt.Println("loopers gonna loop")
|
|
n, err := rin.Read(b)
|
|
if n > 0 {
|
|
rx <- b[:n]
|
|
}
|
|
if nil != err {
|
|
rxErr <- err
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
|
|
for {
|
|
//fmt.Println("poopers gonna poop")
|
|
select {
|
|
// TODO, do we actually need ctx here?
|
|
// would it be sufficient to expect the reader to be closed by the caller instead?
|
|
case <-enc.ctx.Done():
|
|
// TODO: verify that closing the reader will cause the goroutine to be released
|
|
//rin.Close()
|
|
return errors.New("cancelled by context")
|
|
case <-enc.subctx.Done():
|
|
//rin.Close()
|
|
return errors.New("cancelled by context")
|
|
case b := <-rx:
|
|
header, _, err := Encode(b, src, Addr{scheme: src.scheme, addr: "", port: -1})
|
|
if nil != err {
|
|
//rin.Close()
|
|
return err
|
|
}
|
|
//fmt.Println("[debug] encode header:", string(header))
|
|
//fmt.Println("[debug] encode payload:", string(b))
|
|
|
|
_, err = enc.write(header, b)
|
|
if nil != err {
|
|
//rin.Close()
|
|
return err
|
|
}
|
|
case err := <-rxErr:
|
|
// it can be assumed that err will close though, right?
|
|
//rin.Close()
|
|
if io.EOF == err {
|
|
header, _, _ := Encode(nil, src, Addr{scheme: "end"})
|
|
//fmt.Println("[debug] encode end: ", string(header))
|
|
// ignore err, which may have already closed
|
|
_, _ = enc.write(header, nil)
|
|
return nil
|
|
}
|
|
// TODO transmit message , []byte(err.Error())
|
|
header, _, _ := Encode(nil, src, Addr{scheme: "error"})
|
|
// ignore err, which may have already closed
|
|
_, _ = enc.write(header, nil)
|
|
return err
|
|
}
|
|
|
|
}
|
|
}
|
|
|
|
func (enc *Encoder) write(h, b []byte) (int, error) {
|
|
// mutex here so that we can get back error info
|
|
enc.mux.Lock()
|
|
var m int
|
|
n, err := enc.out.Write(append(h, b...))
|
|
/*
|
|
n, err := enc.out.Write(h)
|
|
if nil == err && len(b) > 0 {
|
|
m, err = enc.out.Write(b)
|
|
}
|
|
*/
|
|
enc.mux.Unlock()
|
|
if nil != err {
|
|
enc.outErr <- err
|
|
}
|
|
return n + m, err
|
|
}
|