framestream

README

go-framestream

Frame Streams implementation in Go with compression support (gzip, lz4, snappy and zstd).

Installation

go get -u github.com/dmachard/go-framestream

Usage example

Example to use the framestream library with Net.Pipe

client, server := net.Pipe()
handshake := true

// init framestream sender
go func() {
    fs_server := NewFstrm(bufio.NewReader(server), bufio.NewWriter(server), server, 5*time.Second, []byte("frstrm"), handshake)
    if err := fs_server.InitSender(); err != nil {
        t.Errorf("error to init framestream sender: %s", err)
    }

    // send frame
    frame := &Frame{}
    if err := frame.Write([]byte{1, 2, 3, 4}); err != nil {
        t.Errorf("error to init frame: %s", err)
    }
    if err := fs_server.SendFrame(frame); err != nil {
        t.Errorf("error to send frame: %s", err)
    }
}()

// init framestream receiver
fs_client := NewFstrm(bufio.NewReader(client), bufio.NewWriter(client), client, 5*time.Second, []byte("frstrm"), handshake)
if err := fs_client.InitReceiver(); err != nil {
    t.Errorf("error to init framestream receiver: %s", err)
}

// receive frame, timeout 5s
_, err := fs_client.RecvFrame(true)
if err != nil {
    t.Errorf("error to receive frame: %s", err)
}

Usage example with compression

if err := fs_server.SendCompressedFrame(&compress.GzipCodec, frame); err != nil {
    t.Errorf("error to send frame: %s", err)
}
...
// receive frame, timeout 5s
frame, err := fs_client.RecvCompressedFrame(&compress.GzipCodec, true)
if err != nil {
    t.Errorf("error to receive frame: %s", err)
}

Testing

$ go test -v
=== RUN   TestControlEncode
--- PASS: TestControlEncode (0.00s)
=== RUN   TestControlDecode
--- PASS: TestControlDecode (0.00s)
=== RUN   TestControlDecodeError
--- PASS: TestControlDecodeError (0.00s)
=== RUN   TestFrameWrite
--- PASS: TestFrameWrite (0.00s)
=== RUN   TestFramestreamHandshake
--- PASS: TestFramestreamHandshake (0.00s)
=== RUN   TestFramestreamData
--- PASS: TestFramestreamData (0.00s)
PASS
ok      github.com/dmachard/go-framestream

Frame format

Data Frame

Headers	Bytes
Data length	4 bytes
Payload	xx bytes

Control Frame

Headers	Bytes
Control frame length	4 bytes
Control frame type	4 bytes
Control frame content type	4 bytes (optional)
Control frame content type length	4 bytes (optional)
Content type payload	xx bytes

Documentation

Index

• Constants
• Variables
• type ControlFrame
  • func (ctrl *ControlFrame) CheckContentType(ctype []byte) bool
  • func (ctrl *ControlFrame) Decode() error
  • func (ctrl *ControlFrame) Encode() error
• type Frame
  • func (frame *Frame) AppendData(payload []byte) error
  • func (frame Frame) Data() []byte
  • func (frame *Frame) Encode() error
  • func (frame Frame) IsControl() bool
  • func (frame Frame) Len() int
  • func (frame *Frame) Write(payload []byte) error
• type Fstrm
  • func NewFstrm(reader *bufio.Reader, writer *bufio.Writer, conn net.Conn, ...) *Fstrm
  • func (fs Fstrm) InitReceiver() error
  • func (fs Fstrm) InitSender() error
  • func (fs Fstrm) ProcessFrame(ch chan []byte) error
  • func (fs Fstrm) RecvCompressedFrame(codec compress.Codec, timeout bool) (*Frame, error)
  • func (fs Fstrm) RecvControl() (*ControlFrame, error)
  • func (fs Fstrm) RecvFrame(timeout bool) (*Frame, error)
  • func (fs Fstrm) ResetReceiver(frame *Frame) error
  • func (fs Fstrm) ResetSender() error
  • func (fs Fstrm) SendCompressedFrame(codec compress.Codec, frame *Frame) (err error)
  • func (fs Fstrm) SendControl(control *ControlFrame) (err error)
  • func (fs Fstrm) SendFrame(frame *Frame) (err error)

Constants

const CONTROL_ACCEPT = 0x01

const CONTROL_FIELD_CONTENT_TYPE = 0x01

const CONTROL_FINISH = 0x05

const CONTROL_FRAME_LENGTH_MAX = 4064

const CONTROL_READY = 0x04

const CONTROL_START = 0x02

const CONTROL_STOP = 0x03

const DATA_FRAME_LENGTH_MAX = 65536

Variables

var ErrControlFrameContentTypeUnsupported = errors.New("control frame with unsupported content type")

var ErrControlFrameExpected = errors.New("control frame expected")

var ErrControlFrameMalformed = errors.New("control frame malformed")

var ErrControlFrameTooLarge = errors.New("control frame too large error")

var ErrControlFrameUnexpected = errors.New("control frame unexpected")

var ErrControlFrameUnsupported = errors.New("control frame unsupported")

var ErrFrameTooLarge = errors.New("frame too large error")

var ErrReaderNotReady = errors.New("reader not ready")

Types

type ControlFrame

type ControlFrame struct {
	// contains filtered or unexported fields
}

Control Frame struct

|------------------------------------|----------------------| | Control frame length | 4 bytes | |------------------------------------|----------------------| | Control frame type | 4 bytes | |------------------------------------|----------------------| | Control frame content type | 4 bytes (optional) | |------------------------------------|----------------------| | Control frame content type length | 4 bytes (optional) | |------------------------------------|----------------------| | Content type payload | xx bytes | |------------------------------------|----------------------|

func (*ControlFrame) CheckContentType

func (ctrl *ControlFrame) CheckContentType(ctype []byte) bool

func (*ControlFrame) Decode

func (ctrl *ControlFrame) Decode() error

func (*ControlFrame) Encode

func (ctrl *ControlFrame) Encode() error

type Frame

type Frame struct {
	// contains filtered or unexported fields
}

Frame struct

|------------------------------------|----------------------| | Data length | 4 bytes | |------------------------------------|----------------------| | Payload | xx bytes | |------------------------------------|----------------------|

If the data length is equal to zero then it's a control frame otherwise we have a data frame.

func (*Frame) AppendData

func (frame *Frame) AppendData(payload []byte) error

func (Frame) Data

func (frame Frame) Data() []byte

func (*Frame) Encode

func (frame *Frame) Encode() error

func (Frame) IsControl

func (frame Frame) IsControl() bool

func (Frame) Len

func (frame Frame) Len() int

func (*Frame) Write

func (frame *Frame) Write(payload []byte) error

type Fstrm

type Fstrm struct {
	// contains filtered or unexported fields
}

Framestream

func NewFstrm

func NewFstrm(reader *bufio.Reader, writer *bufio.Writer, conn net.Conn, readtimeout time.Duration, ctype []byte, handshake bool) *Fstrm

func (Fstrm) InitReceiver

func (fs Fstrm) InitReceiver() error

func (Fstrm) InitSender

func (fs Fstrm) InitSender() error

func (Fstrm) ProcessFrame

func (fs Fstrm) ProcessFrame(ch chan []byte) error

func (Fstrm) RecvCompressedFrame

func (fs Fstrm) RecvCompressedFrame(codec compress.Codec, timeout bool) (*Frame, error)

func (Fstrm) RecvControl

func (fs Fstrm) RecvControl() (*ControlFrame, error)

func (Fstrm) RecvFrame

func (fs Fstrm) RecvFrame(timeout bool) (*Frame, error)

func (Fstrm) ResetReceiver

func (fs Fstrm) ResetReceiver(frame *Frame) error

func (Fstrm) ResetSender

func (fs Fstrm) ResetSender() error

func (Fstrm) SendCompressedFrame

func (fs Fstrm) SendCompressedFrame(codec compress.Codec, frame *Frame) (err error)

func (Fstrm) SendControl

func (fs Fstrm) SendControl(control *ControlFrame) (err error)

func (Fstrm) SendFrame

func (fs Fstrm) SendFrame(frame *Frame) (err error)