pipes

README

pipes

Pipes is a framework we use at Fajurion to handle real-time messaging for our chat app. It works across multiple nodes with multiple servers for ultimate scalability. It currently works with UDP and WebSockets. There is still some need for performance improvements in the future, but it will be fine for small applications.

There is also no documentation as of writing this, there'll eventually be a documentation page for this API, but that'll happen when it is fully released.

NOTE: Pipes is currently in early stages, I'm not sure how I'll design the final API when it's released: All API features are subject to change.

Documentation

Index

• Constants
• Variables
• type Adapter
• type AdoptionRequest
• type Channel
  • func BroadcastChannel(receivers []string) Channel
  • func Conversation(receivers []string, nodes []string) Channel
  • func P2PChannel(receiver string, receiverNode string) Channel
  • func (c Channel) IsBroadcast() bool
  • func (c Channel) IsConversation() bool
  • func (c Channel) IsP2P() bool
• type Context
• type Event
• type LocalNode
  • func SetupCurrent(id string, token string) *LocalNode
  • func (node *LocalNode) AdaptWS(adapter Adapter)
  • func (node *LocalNode) AdapterReceiveWeb(ID string, event Event, msg []byte) error
  • func (local *LocalNode) AddNode(node Node)
  • func (local *LocalNode) ConnectToNodeWS(node Node) error
  • func (local *LocalNode) Decrypt(node string, msg []byte) ([]byte, error)
  • func (local *LocalNode) DeleteNode(node string)
  • func (local *LocalNode) Encrypt(node string, msg []byte) ([]byte, error)
  • func (local *LocalNode) ExistsNodeWS(node string) bool
  • func (local *LocalNode) GetNode(id string) *Node
  • func (local *LocalNode) GetNodeWS(node string) *websocket.Conn
  • func (local *LocalNode) HandleMessage(protocol string, message Message)
  • func (local *LocalNode) IterateNodes(callback func(string, Node) bool)
  • func (local *LocalNode) IterateNodesWS(f func(key string, value *websocket.Conn) bool)
  • func (node *LocalNode) Pipe(protocol string, message Message) error
  • func (local *LocalNode) ProcessEvent(message *Message, target string) Event
  • func (local *LocalNode) ProcessMarshal(message *Message, target string) []byte
  • func (local *LocalNode) ReceiveWS(bytes []byte)
  • func (local *LocalNode) ReceiveWSAdoption(request string) (Node, error)
  • func (node *LocalNode) RemoveAdapterWS(ID string)
  • func (local *LocalNode) RemoveNodeWS(node string)
  • func (local *LocalNode) SendClient(id string, event Event) error
  • func (n *LocalNode) SetupSocketless(sl string)
  • func (n *LocalNode) SetupWS(ws string)
  • func (local *LocalNode) Socketless(nodeEntity Node, message Message) error
  • func (local *LocalNode) ToNode() Node
• type Message
• type Node

Constants

const ChannelBroadcast = "br"

const ChannelConversation = "c"

const ChannelP2P = "p"

const ProtocolUDP = "udp"

const ProtocolWS = "ws"

Variables

var DebugLogs = false

var Log = log.New(log.Writer(), "pipes ", log.Flags())

Types

type Adapter

type Adapter struct {
	ID    string      // Identifier of the client
	Mutex *sync.Mutex // Mutex to prevent concurrent sending (WHY DO I NEED TO DO THIS??)
	Data  interface{} // Custom data (not required)

	// Functions
	Receive func(*Context) error
	OnError func(error)
}

type AdoptionRequest

type AdoptionRequest struct {
	Token    string `json:"tk"`
	Adopting Node   `json:"adpt"`
}

type Channel

type Channel struct {
	Channel string   `json:"channel"` // Channel name
	Target  []string `json:"target"`  // User IDs to send to (node and user ID for p2p channel)
	Nodes   []string `json:"-"`       // Nodes to send to (only for conversation channel)
}

func BroadcastChannel

func BroadcastChannel(receivers []string) Channel

func Conversation

func Conversation(receivers []string, nodes []string) Channel

func P2PChannel

func P2PChannel(receiver string, receiverNode string) Channel

func (Channel) IsBroadcast

func (c Channel) IsBroadcast() bool

func (Channel) IsConversation

func (c Channel) IsConversation() bool

func (Channel) IsP2P

func (c Channel) IsP2P() bool

type Context

type Context struct {
	Event   *Event
	Message []byte
	Adapter *Adapter
}

type Event

type Event struct {
	Name string                 `json:"name"`
	Data map[string]interface{} `json:"data"`
}

type LocalNode

type LocalNode struct {
	ID    string `json:"id"`
	Token string `json:"token"`
	WS    string `json:"ws,omitempty"` // Websocket ip
	SL    string `json:"sl,omitempty"` // Socketless pipe

	Processors map[string]func(*Message, string) Event `json:"-"`

	// Encryption
	Cipher cipher.Block `json:"-"`
	// contains filtered or unexported fields
}

func SetupCurrent

func SetupCurrent(id string, token string) *LocalNode

func (*LocalNode) AdaptWS

func (node *LocalNode) AdaptWS(adapter Adapter)

Register a new adapter for websocket/sl (all safe protocols)

func (*LocalNode) AdapterReceiveWeb

func (node *LocalNode) AdapterReceiveWeb(ID string, event Event, msg []byte) error

Handles receiving messages from the target and passes them to the adapter

func (*LocalNode) AddNode

func (local *LocalNode) AddNode(node Node)

func (*LocalNode) ConnectToNodeWS

func (local *LocalNode) ConnectToNodeWS(node Node) error

func (*LocalNode) Decrypt

func (local *LocalNode) Decrypt(node string, msg []byte) ([]byte, error)

func (*LocalNode) DeleteNode

func (local *LocalNode) DeleteNode(node string)

func (*LocalNode) Encrypt

func (local *LocalNode) Encrypt(node string, msg []byte) ([]byte, error)

func (*LocalNode) ExistsNodeWS

func (local *LocalNode) ExistsNodeWS(node string) bool

func (*LocalNode) GetNode

func (local *LocalNode) GetNode(id string) *Node

func (*LocalNode) GetNodeWS

func (local *LocalNode) GetNodeWS(node string) *websocket.Conn

func (*LocalNode) HandleMessage

func (local *LocalNode) HandleMessage(protocol string, message Message)

func (*LocalNode) IterateNodes

func (local *LocalNode) IterateNodes(callback func(string, Node) bool)

IterateConnections iterates over all connections. If the callback returns false, the iteration stops.

func (*LocalNode) IterateNodesWS

func (local *LocalNode) IterateNodesWS(f func(key string, value *websocket.Conn) bool)

Range calls f sequentially for each key and value present in the map. If f returns false, range stops the iteration.

func (*LocalNode) Pipe

func (node *LocalNode) Pipe(protocol string, message Message) error

func (*LocalNode) ProcessEvent

func (local *LocalNode) ProcessEvent(message *Message, target string) Event

func (*LocalNode) ProcessMarshal

func (local *LocalNode) ProcessMarshal(message *Message, target string) []byte

func (*LocalNode) ReceiveWS

func (local *LocalNode) ReceiveWS(bytes []byte)

func (*LocalNode) ReceiveWSAdoption

func (local *LocalNode) ReceiveWSAdoption(request string) (Node, error)

func (*LocalNode) RemoveAdapterWS

func (node *LocalNode) RemoveAdapterWS(ID string)

Remove a websocket/sl adapter

func (*LocalNode) RemoveNodeWS

func (local *LocalNode) RemoveNodeWS(node string)

func (*LocalNode) SendClient

func (local *LocalNode) SendClient(id string, event Event) error

SendClient is a function that sends a WS packet to the client

func (*LocalNode) SetupSocketless

func (n *LocalNode) SetupSocketless(sl string)

func (*LocalNode) SetupWS

func (n *LocalNode) SetupWS(ws string)

func (*LocalNode) Socketless

func (local *LocalNode) Socketless(nodeEntity Node, message Message) error

func (*LocalNode) ToNode

func (local *LocalNode) ToNode() Node

type Message

type Message struct {
	Channel Channel `json:"channel"`
	Event   Event   `json:"event"`
	Local   bool    `json:"-"` // Whether to only send to local clients (excluded from JSON)
}

type Node

type Node struct {
	ID    string `json:"id"`
	Token string `json:"token"`
	WS    string `json:"ws,omitempty"` // Websocket ip
	SL    string `json:"sl,omitempty"` // Socketless pipe

	// Encryption
	Cipher cipher.Block `json:"-"`
}