goetty

README

goetty

Goetty is a framework to help you build socket application.

Example

server

package example

import (
	"log"

	"github.com/fagongzi/goetty"
	"github.com/fagongzi/goetty/codec/simple"
)

// EchoServer echo server
type EchoServer struct {
	addr string
	app  goetty.NetApplication
}

// NewEchoServer create new server
func NewEchoServer(addr string) *EchoServer {
	svr := &EchoServer{}
	encoder, decoder := simple.NewStringCodec()
	app, err := goetty.NewTCPApplication(addr, svr.handle,
		goetty.WithAppSessionOptions(goetty.WithCodec(encoder, decoder)))
	if err != nil {
		log.Panicf("start server failed with %+v", err)
	}

	return &EchoServer{
		addr: addr,
		app:  app,
	}
}

// Start start
func (s *EchoServer) Start() error {
	return s.Start()
}

// Stop stop
func (s *EchoServer) Stop() error {
	return s.Stop()
}

func (s *EchoServer) handle(session goetty.IOSession, msg interface{}, received uint64) error {
	log.Printf("received %+v from %s, already received %d msgs",
		msg,
		session.RemoteAddr(),
		received)
	return session.WriteAndFlush(msg)
}

client

package example

import (
	"log"
	"time"

	"github.com/fagongzi/goetty"
	"github.com/fagongzi/goetty/codec/simple"
)

// EchoClient echo client
type EchoClient struct {
	serverAddr string
	conn       goetty.IOSession
}

// NewEchoClient new client
func NewEchoClient(serverAddr string) (*EchoClient, error) {
	c := &EchoClient{
		serverAddr: serverAddr,
	}

	encoder, decoder := simple.NewStringCodec()
	c.conn = goetty.NewIOSession(goetty.WithCodec(encoder, decoder))
	_, err := c.conn.Connect(serverAddr, time.Second*3)
	return c, err
}

// SendMsg send msg to server
func (c *EchoClient) SendMsg(msg string) error {
	return c.conn.WriteAndFlush(msg)
}

// ReadLoop read loop
func (c *EchoClient) ReadLoop() error {
	// start loop to read msg from server
	for {
		msg, err := c.conn.Read() // if you want set a read deadline, you can use 'WithTimeout option'
		if err != nil {
			log.Printf("read failed with %+v", err)
			return err
		}

		log.Printf("received %+v", msg)
	}
}

Documentation

Index

• Variables
• type AppOption
  • func WithAppHandleSessionFunc[IN any, OUT any](value func(IOSession[IN, OUT]) error) AppOption[IN, OUT]
  • func WithAppLogger[IN any, OUT any](logger *zap.Logger) AppOption[IN, OUT]
  • func WithAppSessionAware[IN any, OUT any](value IOSessionAware[IN, OUT]) AppOption[IN, OUT]
  • func WithAppSessionBucketSize[IN any, OUT any](value uint64) AppOption[IN, OUT]
  • func WithAppSessionOptions[IN any, OUT any](options ...Option[IN, OUT]) AppOption[IN, OUT]
  • func WithAppTLS[IN any, OUT any](tlsCfg *tls.Config) AppOption[IN, OUT]
  • func WithAppTLSFromCertAndKey[IN any, OUT any](certFile string, keyFile string, caFile string, insecureSkipVerify bool) AppOption[IN, OUT]
• type BufferedIOSession
• type IOSession
  • func NewIOSession[IN any, OUT any](opts ...Option[IN, OUT]) IOSession[IN, OUT]
• type IOSessionAware
• type NetApplication
  • func NewApplication[IN any, OUT any](address string, handleFunc func(IOSession[IN, OUT], IN, uint64) error, ...) (NetApplication[IN, OUT], error)
  • func NewApplicationWithListenAddress[IN any, OUT any](addresses []string, handleFunc func(IOSession[IN, OUT], IN, uint64) error, ...) (NetApplication[IN, OUT], error)
  • func NewApplicationWithListeners[IN any, OUT any](listeners []net.Listener, ...) (NetApplication[IN, OUT], error)
• type Option
  • func WithSessionAllocator[IN any, OUT any](allocator buf.Allocator) Option[IN, OUT]
  • func WithSessionAware[IN any, OUT any](value IOSessionAware[IN, OUT]) Option[IN, OUT]
  • func WithSessionCodec[IN any, OUT any](codec codec.Codec[IN, OUT]) Option[IN, OUT]
  • func WithSessionConn[IN any, OUT any](id uint64, conn net.Conn) Option[IN, OUT]
  • func WithSessionDisableAutoResetInBuffer[IN any, OUT any]() Option[IN, OUT]
  • func WithSessionDisableCompactAfterGrow[IN any, OUT any]() Option[IN, OUT]
  • func WithSessionLogger[IN any, OUT any](logger *zap.Logger) Option[IN, OUT]
  • func WithSessionRWBufferSize[IN any, OUT any](read, write int) Option[IN, OUT]
  • func WithSessionReleaseMsgFunc[IN any, OUT any](value func(any)) Option[IN, OUT]
  • func WithSessionTLS[IN any, OUT any](tlsConfig *tls.Config) Option[IN, OUT]
  • func WithSessionTLSFromCertAndKeys[IN any, OUT any](certFile, keyFile, caFile string, insecureSkipVerify bool) Option[IN, OUT]
• type Proxy
  • func NewProxy[IN any, OUT any](address string, logger *zap.Logger) Proxy
• type ReadOptions
• type WriteOptions

Variables

var (
	// ErrIllegalState illegal state error
	ErrIllegalState = errors.New("illegal state")
	// ErrDisableConnect disable to connect
	ErrDisableConnect = errors.New("io session is disable to connect")
)

Types

type AppOption

type AppOption[IN any, OUT any] func(*server[IN, OUT])

AppOption application option

func WithAppHandleSessionFunc

func WithAppHandleSessionFunc[IN any, OUT any](value func(IOSession[IN, OUT]) error) AppOption[IN, OUT]

WithAppHandleSessionFunc set the app handle session func

func WithAppLogger

func WithAppLogger[IN any, OUT any](logger *zap.Logger) AppOption[IN, OUT]

WithAppLogger set logger for application

func WithAppSessionAware

func WithAppSessionAware[IN any, OUT any](value IOSessionAware[IN, OUT]) AppOption[IN, OUT]

WithAppSessionBucketSize set the app session aware

func WithAppSessionBucketSize

func WithAppSessionBucketSize[IN any, OUT any](value uint64) AppOption[IN, OUT]

WithAppSessionBucketSize set the number of maps to store session

func WithAppSessionOptions

func WithAppSessionOptions[IN any, OUT any](options ...Option[IN, OUT]) AppOption[IN, OUT]

WithAppSessionOptions set options to create new connection

func WithAppTLS

func WithAppTLS[IN any, OUT any](tlsCfg *tls.Config) AppOption[IN, OUT]

WithAppTLS set tls config for application

func WithAppTLSFromCertAndKey

func WithAppTLSFromCertAndKey[IN any, OUT any](
	certFile string,
	keyFile string,
	caFile string,
	insecureSkipVerify bool) AppOption[IN, OUT]

WithAppTLSFromKeys set tls config from cert and key files for application

type BufferedIOSession

type BufferedIOSession interface {
	// BufferedConn returns a wrapped net.Conn that read from IOSession's in-buffer first
	BufferedConn() net.Conn
}

BufferedIOSession is a IOSession that can read from the in-buffer first

type IOSession

type IOSession[IN any, OUT any] interface {
	// ID session id
	ID() uint64
	// Connect connect to address, only used at client-side
	Connect(addr string, timeout time.Duration) error
	// Connected returns true if connection is ok
	Connected() bool
	// Disconnect disconnect the connection
	Disconnect() error
	// Close close the session, the read and write buffer will closed, and cannot Connect
	// again. IOSession reference count minus 1.
	Close() error
	// Ref for IOSessions, held by several goroutines, several references are needed. Each
	// concurrent process holding an IOSession can Close the IOSession and release the resource
	// when the reference count reaches 0.
	Ref()
	// Read read packet from connection
	Read(option ReadOptions) (IN, error)
	// Write encodes the msg into a []byte into the buffer according to the codec.Encode.
	// If flush is set to false, the data will not be written to the underlying socket.
	Write(msg OUT, options WriteOptions) error
	// Flush flush the out buffer
	Flush(timeout time.Duration) error
	// RemoteAddress returns remote address, include ip and port
	RemoteAddress() string
	// RawConn return raw tcp conn, RawConn should only be used to access the underlying
	// attributes of the tcp conn, e.g. set keepalive attributes. Read from RawConn directly
	// may lose data since the bytes might have been copied to the InBuf.
	// To perform read/write operation on the underlying tcp conn, use BufferedConn instead.
	RawConn() net.Conn
	// UseConn use the specified conn to handle reads and writes. Note that conn reads and
	// writes cannot be handled in other goroutines until UseConn is called.
	UseConn(net.Conn)
	// OutBuf returns byte buffer which used to encode message into bytes
	OutBuf() *buf.ByteBuf
	// InBuf returns input buffer which used to decode bytes to message
	InBuf() *buf.ByteBuf
}

IOSession internally holds a raw net.Conn on which to provide read and write operations

func NewIOSession

func NewIOSession[IN any, OUT any](opts ...Option[IN, OUT]) IOSession[IN, OUT]

NewIOSession create a new io session

type IOSessionAware

type IOSessionAware[IN any, OUT any] interface {
	// Created session created
	Created(IOSession[IN, OUT])
	//Closed session closed
	Closed(IOSession[IN, OUT])
}

IOSessionAware io session aware

type NetApplication

type NetApplication[IN any, OUT any] interface {
	// Start start the transport server
	Start() error
	// Stop stop the transport server
	Stop() error
	// GetSession get session
	GetSession(uint64) (IOSession[IN, OUT], error)
}

NetApplication is a network based application

func NewApplication

func NewApplication[IN any, OUT any](
	address string,
	handleFunc func(IOSession[IN, OUT], IN, uint64) error,
	opts ...AppOption[IN, OUT]) (NetApplication[IN, OUT], error)

NewApplication returns a application

func NewApplicationWithListenAddress

func NewApplicationWithListenAddress[IN any, OUT any](
	addresses []string,
	handleFunc func(IOSession[IN, OUT], IN, uint64) error,
	opts ...AppOption[IN, OUT]) (NetApplication[IN, OUT], error)

NewApplicationWithListenAddress create a net application with listen multi addresses

func NewApplicationWithListeners

func NewApplicationWithListeners[IN any, OUT any](
	listeners []net.Listener,
	handleFunc func(IOSession[IN, OUT], IN, uint64) error,
	opts ...AppOption[IN, OUT]) (NetApplication[IN, OUT], error)

NewApplicationWithListener returns a net application with listener

type Option

type Option[IN any, OUT any] func(*baseIO[IN, OUT])

Option option to create IOSession

func WithSessionAllocator

func WithSessionAllocator[IN any, OUT any](allocator buf.Allocator) Option[IN, OUT]

WithSessionAllocator set mem allocator to build in and out ByteBuf

func WithSessionAware

func WithSessionAware[IN any, OUT any](value IOSessionAware[IN, OUT]) Option[IN, OUT]

WithSessionAware set IOSession's session aware

func WithSessionCodec

func WithSessionCodec[IN any, OUT any](codec codec.Codec[IN, OUT]) Option[IN, OUT]

WithSessionCodec set codec for IOSession

func WithSessionConn

func WithSessionConn[IN any, OUT any](id uint64, conn net.Conn) Option[IN, OUT]

WithSessionConn set IOSession's net.Conn

func WithSessionDisableAutoResetInBuffer

func WithSessionDisableAutoResetInBuffer[IN any, OUT any]() Option[IN, OUT]

WithSessionDisableAutoResetInBuffer set disable auto reset in buffer. If disabled, the application must reset in buffer in the read loop, otherwise there will be a memory leak.

func WithSessionDisableCompactAfterGrow

func WithSessionDisableCompactAfterGrow[IN any, OUT any]() Option[IN, OUT]

WithSessionDisableCompactAfterGrow set Set whether the buffer should be compressed, if it is, it will reset the reader and writer index. Default is true.

func WithSessionLogger

func WithSessionLogger[IN any, OUT any](logger *zap.Logger) Option[IN, OUT]

WithSessionLogger set logger for IOSession

func WithSessionRWBufferSize

func WithSessionRWBufferSize[IN any, OUT any](read, write int) Option[IN, OUT]

WithSessionRWBufferSize set read/write buf size for IOSession

func WithSessionReleaseMsgFunc

func WithSessionReleaseMsgFunc[IN any, OUT any](value func(any)) Option[IN, OUT]

WithSessionReleaseMsgFunc set a func to release message once the message encode into the write buf

func WithSessionTLS

func WithSessionTLS[IN any, OUT any](tlsConfig *tls.Config) Option[IN, OUT]

WithSessionTLS set tls for client

func WithSessionTLSFromCertAndKeys

func WithSessionTLSFromCertAndKeys[IN any, OUT any](certFile, keyFile, caFile string, insecureSkipVerify bool) Option[IN, OUT]

WithSessionTLSFromCertAndKeys set tls for client

type Proxy

type Proxy interface {
	// Start start the proxy
	Start() error
	// Stop stop the proxy
	Stop() error
	// AddUpStream add upstream
	AddUpStream(address string, connectTimeout time.Duration)
}

Proxy simple reverse proxy

func NewProxy

func NewProxy[IN any, OUT any](address string, logger *zap.Logger) Proxy

NewProxy returns a simple tcp proxy

type ReadOptions

type ReadOptions struct {
	// Timeout deadline for read
	Timeout time.Duration
}

ReadOptions read options

type WriteOptions

type WriteOptions struct {
	// Timeout deadline for write
	Timeout time.Duration
	// Flush flush data to net.Conn
	Flush bool
}

WriteOptions write options