simplesub

README

go-simplesub

A minimalistic, yet powerful pubsub messaging system built on top of libp2p.

Rationale

Why does this repo exist?

Simple: Libp2p's pub/sub implementation simply does not provide enough flexibility for some use cases (those that utilize a DHT or any sort of routing in particular).

Installation

go get github.com/dowlandaiello/go-simplesub

Usage

package main

import (
    "context"

    "github.com/libp2p/go-libp2p"
    dht "github.com/libp2p/go-libp2p-kad-dht"
    routed "github.com/libp2p/go-libp2p/p2p/host/routed"
    "github.com/dowlandaiello/go-simplesub"
    inet "github.com/libp2p/go-libp2p-net"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background()) // Initialize context

    defer cancel() // Cancel

    host, err := libp2p.New(
        ctx,
        libp2p.NATPortMap(),
        libp2p.ListenAddrStrings(
            "/ip4/0.0.0.0/tcp/1111",
            "/ip6/::1/tcp/1111",
        ),
    ) // Initialize host

    if err != nil { // Check for errors
        panic(err) // Panic
    }

    dht, err := dht.New(ctx, host) // Initialize dht

    if err != nil { // Check for errors
        panic(err) // Panic
    }

    routedHost := routed.Wrap(host, dht) // Wrap host

    sub, err := simplesub.NewSimpleSub(routedHost) // Initialize sub

    if err != nil { // Check for errors
        panic(err) // Panic
    }

    sub.Subscribe("test_topic", handler) // Subscribe

    err = sub.Publish(ctx, "test_topic", []byte("test")) // Publish to topic

    if err != nil { // Check for errors
        panic(err) // Panic
    }
}

// handler handles a new incoming stream.
func handler(stream inet.Stream, message *simplesub.Message) {
    fmt.Printf("Received message: %s", string(message.Data)) // Log received
}

Configuration

Route Prefixes

By default, all simplesub routes are registered under /. Should one wish to add a prefix to such a route, simply pass the WithRoutePrefix option function to the simplesub constructor.

Example:

sub, err := simplesub.NewSimpleSub(routedHost, simplesub.WithRoutePrefix("test_net")) // Initialize sub

if err != nil { // Check for errors
    panic(err) // Panic
}

Documentation

Overview

Package simplesub implements a pub/sub messaging system through the libp2p routed.RoutedHost interface. In contrast with the standard libp2p pub/sub package, simplesub has the advantage of letting developers opt for their own routing solutions (e.g. kadDHT).

Package simplesub implements a pub/sub messaging system through the libp2p routed.RoutedHost interface. In contrast with the standard libp2p pub/sub package, simplesub has the advantage of letting developers opt for their own routing solutions (e.g. kadDHT).

Package simplesub implements a pub/sub messaging system through the libp2p routed.RoutedHost interface. In contrast with the standard libp2p pub/sub package, simplesub has the advantage of letting developers opt for their own routing solutions (e.g. kadDHT).

Package simplesub implements a pub/sub messaging system through the libp2p routed.RoutedHost interface. In contrast with the standard libp2p pub/sub package, simplesub has the advantage of letting developers opt for their own routing solutions (e.g. kadDHT).

Index

• type Message
  • func MessageFromBytes(b []byte) (*Message, error)
  • func (message *Message) Bytes() ([]byte, error)
• type Option
  • func WithRoutePrefix(prefix string) Option
• type SimpleSub
  • func NewSimpleSub(host *routed.RoutedHost, opts ...Option) (*SimpleSub, error)
  • func (sub *SimpleSub) Publish(ctx context.Context, topic string, data []byte, peers ...peer.ID) error
  • func (sub *SimpleSub) Subscribe(topic string, handler func(inet.Stream, *Message))

Types

type Message

type Message struct {
	Topic string `json:"topic"` // Message topic

	Data []byte `json:"data"` // Message data
}

Message outlines a simplesub message.

func MessageFromBytes

func MessageFromBytes(b []byte) (*Message, error)

MessageFromBytes attempts to decode a message from a given byte slice.

func (*Message) Bytes

func (message *Message) Bytes() ([]byte, error)

Bytes serializes a given message to a byte slice.

type Option

type Option func(*SimpleSub) error

Option represents a simplesub configuration option.

func WithRoutePrefix

func WithRoutePrefix(prefix string) Option

WithRoutePrefix defines the WithRoutePrefix option. Such an option can used primarily to differentiate between different nodes in a network, or partition such networks.

type SimpleSub

type SimpleSub struct {
	Host *routed.RoutedHost `json:"host"` // Working host

	RootRoutePath string `json:"root_path"` // Root route path

	Handlers map[string]func(inet.Stream, *Message) // Message handlers
}

SimpleSub implements the standard simplesub pub/sub messaging system. The simplesub type is in many ways analogous to the host type in libp2p--it serves as a central hub for all pub/sub related operations.

func NewSimpleSub

func NewSimpleSub(host *routed.RoutedHost, opts ...Option) (*SimpleSub, error)

NewSimpleSub initializes a new SimpleSub, and sets up all necessary stream handlers.

func (*SimpleSub) Publish

func (sub *SimpleSub) Publish(ctx context.Context, topic string, data []byte, peers ...peer.ID) error

Publish publishes to a given topic, to a given subset of peers. If no target peers are specified, the message is broadcasted to the entire network (i.e. all peers).

func (*SimpleSub) Subscribe

func (sub *SimpleSub) Subscribe(topic string, handler func(inet.Stream, *Message))

Subscribe subscribes to a given topic.