func (sub *Subscriber) SubscribeToAll() (chan []byte, error) {
	sub.chanLock.Lock()
	defer sub.chanLock.Unlock()
	if sub.allOutput == nil {
		allOutput := make(chan []byte, channelBuffer)
		err := sub.sock.SetOption(mangos.OptionSubscribe, []byte(""))
		if err != nil {
			return nil, fmt.Errorf("set subscribe socket option: %w", err)
		}
		sub.allOutput = allOutput
	}
	return sub.allOutput, nil
}
SubscribeToAll subscribes to all available topics.
package events
import (
	"fmt"
	"sync"
	"time"
	"nanomsg.org/go-mangos"
	"nanomsg.org/go-mangos/protocol/pub"
	"nanomsg.org/go-mangos/protocol/sub"
	"nanomsg.org/go-mangos/transport/ipc"
	"nanomsg.org/go-mangos/transport/tcp"
	"github.com/spacemeshos/go-spacemesh/log"
)
const channelBuffer = 100
type ChannelID byte
type Subscriber struct {
	sock   mangos.Socket
	closer chan struct{}
	chanLock  sync.RWMutex
	output    map[ChannelID]chan []byte
	allOutput chan []byte
}
func NewSubscriber(url string) (*Subscriber, error) {
	socket, err := sub.NewSocket()
	if err != nil {
		return nil, fmt.Errorf("create socket: %w", err)
	}
	socket.AddTransport(ipc.NewTransport())
	socket.AddTransport(tcp.NewTransport())
	socket.SetOption(mangos.OptionBestEffort, false)
	socket.SetOption(mangos.OptionReadQLen, 10000)
	err = socket.Dial(url)
	if err != nil {
		return nil, fmt.Errorf("dial socket: %w", err)
	}
	return &Subscriber{
		sock:   socket,
		output: make(map[ChannelID]chan []byte, 100),
		closer: make(chan struct{}),
	}, nil
}
func (sub *Subscriber) StartListening() {
	go func() {
		for {
			payload, err := sub.sock.Recv()
			log.With().Debug("got message", log.Binary("payload", payload))
			if err != nil {
				if err.Error() == "connection closed" {
					log.Warning("connection closed on pubsub reader")
				} else {
					log.Error("error on recv: %v", err)
				}
				return
			}
			sub.route(payload)
		}
	}()
}
func (sub *Subscriber) route(payload []byte) {
	if len(payload) < 1 {
		log.Error("got empty message")
		return
	}
	sub.chanLock.RLock()
	defer sub.chanLock.RUnlock()
	var receiver chan []byte
	if sub.allOutput != nil {
		receiver = sub.allOutput
	} else {
		if c, ok := sub.output[ChannelID(payload[0])]; ok {
			receiver = c
		}
	}
	if receiver == nil {
		return
	}
	select {
	case receiver <- payload:
	case <-sub.closer:
	}
}
func (sub *Subscriber) Close() error {
	select {
	case <-sub.closer:
	default:
		close(sub.closer)
		if err := sub.sock.Close(); err != nil {
			return fmt.Errorf("close socket: %w", err)
		}
		return nil
	}
	return nil
}
func (sub *Subscriber) Subscribe(topic ChannelID) (chan []byte, error) {
	sub.chanLock.Lock()
	defer sub.chanLock.Unlock()
	if _, ok := sub.output[topic]; !ok {
		sub.output[topic] = make(chan []byte, channelBuffer)
	}
	err := sub.sock.SetOption(mangos.OptionSubscribe, []byte{byte(topic)})
	if err != nil {
		return nil, fmt.Errorf("set subscribe socket option: %w", err)
	}
	return sub.output[topic], nil
}