etp

README

isp-etp-go

Client and server implementation of event transport protocol. Websocket framework based on library github.com/nhooyr/websocket.

Install

go get github.com/integration-system/isp-etp-go

Features:

• Rooms, broadcasting
• Store data object for connection
• Javascript client
• Concurrent message write
• Context based requests

Server example

package main

import (
	"context"
	"errors"
	"github.com/integration-system/isp-etp-go"
	"log"
	"net/http"
	"nhooyr.io/websocket"
)

func main() {
	bindAddress := "127.0.0.1:7777"
	testEvent := "test_event"
	helloEvent := "hello"
	unknownEvent := "unknown"

	config := etp.ServerConfig{
		InsecureSkipVerify: true,
	}
	server := etp.NewServer(context.TODO(), config).
		OnConnect(func(conn etp.Conn) {
			log.Println("OnConnect", conn.ID())
			err := conn.Emit(context.TODO(), unknownEvent, []byte("qwerty"))
			log.Println("unknown answer err:", err)
		}).
		OnDisconnect(func(conn etp.Conn, err error) {
			log.Println("OnDisconnect id", conn.ID())
			log.Println("OnDisconnect err:", err)
			var closeErr websocket.CloseError
			if errors.As(err, &closeErr) {
				log.Println("OnDisconnect close code:", closeErr.Code)
			}
		}).
		OnError(func(conn etp.Conn, err error) {
			log.Println("OnError err:", err)
			if conn != nil {
				log.Println("OnError conn ID:", conn.ID())
			}
		}).
		On(testEvent, func(conn etp.Conn, data []byte) {
			log.Println("Received " + testEvent + ":" + string(data))
		}).
		On(helloEvent, func(conn etp.Conn, data []byte) {
			log.Println("Received " + helloEvent + ":" + string(data))
			answer := "hello, " + conn.ID()
			err := conn.Emit(context.TODO(), testEvent, []byte(answer))
			log.Println("hello answer err:", err)
		})
	// used as alternative to wildcards and custom handling
	server.OnDefault(func(event string, conn etp.Conn, data []byte) {
		log.Printf("Received default %s:%s\n", event, string(data))
	})

	mux := http.NewServeMux()
	mux.HandleFunc("/isp-etp/", server.ServeHttp)
	httpServer := &http.Server{Addr: bindAddress, Handler: mux}
	go func() {
		if err := httpServer.ListenAndServe(); err != nil {
			log.Fatalf("Unable to start http server. err: %v", err)
		}
	}()

	<-make(chan struct{})
}

Client example

package main

import (
	"context"
	"errors"
	etpclient "github.com/integration-system/isp-etp-go/client"
	"log"
	"net/http"
	"nhooyr.io/websocket"
)

func main() {
	address := "ws://127.0.0.1:7777/isp-etp/"
	testEvent := "test_event"
	helloEvent := "hello"
	config := etpclient.Config{
		HttpClient: http.DefaultClient,
	}
	client := etpclient.NewClient(config).
		OnConnect(func() {
			log.Println("Connected")
		}).
		OnDisconnect(func(err error) {
			log.Println("OnDisconnect err:", err)
			var closeErr websocket.CloseError
			if errors.As(err, &closeErr) {
				log.Println("OnDisconnect close code:", closeErr.Code)
			}
		}).
		OnError(func(err error) {
			log.Println("OnError err:", err)
		})
	client.On(testEvent, func(data []byte) {
		log.Printf("Received %s:%s\n", testEvent, string(data))
	})
	// used as alternative to wildcards and custom handling
	client.OnDefault(func(event string, data []byte) {
		log.Printf("Received default %s:%s\n", event, string(data))
	})
	err := client.Dial(context.TODO(), address)
	if err != nil {
		log.Fatalln("dial error:", err)
	}

	err = client.Emit(context.TODO(), helloEvent, []byte("hello"))
	if err != nil {
		log.Fatalln("emit error:", err)
	}

	<-make(chan struct{})
}

Documentation

Index

• type Conn
• type RoomStore
  • func NewRoomStore() RoomStore
• type Server
  • func NewServer(ctx context.Context, config ServerConfig) Server
• type ServerConfig

Types

type Conn

type Conn interface {
	ID() string
	Close() error
	URL() *url.URL
	RemoteAddr() string
	RemoteHeader() http.Header
	Context() interface{}
	SetContext(v interface{})
	Emit(ctx context.Context, event string, body []byte) error
	EmitWithAck(ctx context.Context, event string, body []byte) ([]byte, error)
	Closed() bool
}

type RoomStore

type RoomStore interface {
	Add(Conn)
	Remove(Conn)
	Get(connId string) (Conn, bool)
	Join(conn Conn, rooms ...string)
	Leave(conn Conn, rooms ...string)
	LeaveByConnId(id string, rooms ...string)
	Len(room string) int
	Clear(rooms ...string)
	Rooms() []string
	ToBroadcast(rooms ...string) []Conn
}

func NewRoomStore

func NewRoomStore() RoomStore

type Server

type Server interface {
	Close()
	ServeHttp(w http.ResponseWriter, r *http.Request)
	OnWithAck(event string, f func(conn Conn, data []byte) []byte) Server
	// If registered, all unknown events will be handled here.
	OnDefault(f func(event string, conn Conn, data []byte)) Server
	On(event string, f func(conn Conn, data []byte)) Server
	Unsubscribe(event string) Server
	OnConnect(f func(Conn)) Server
	OnDisconnect(f func(Conn, error)) Server
	// Conn may be nil if error occurs on connection upgrade or in RequestHandler.
	OnError(f func(Conn, error)) Server

	Rooms() RoomStore
	BroadcastToRoom(room string, event string, data []byte) error
	BroadcastToAll(event string, data []byte) error
}

func NewServer

func NewServer(ctx context.Context, config ServerConfig) Server

type ServerConfig

type ServerConfig struct {
	// Can be used to define custom CORS policy.
	RequestHandler func(*http.Request) error
	// By default, the connection has a message read limit of 32768 bytes.
	// When the limit is hit, the connection will be closed with StatusMessageTooBig.
	ConnectionReadLimit int64
	// Checks origin header to prevent CSRF attack.
	InsecureSkipVerify bool
}