func (ep *Endpoint) handleMessage(msg [][]byte) {
	appName := string(msg[0])
	switch {
	case len(msg) < 4:
		log.Warnf("zmq3<PubSub>: Invalid message received from %s: message too short", appName)
		return
	case !bytes.Equal(msg[2], FrameHeader):
		log.Warnf("zmq3<PubSub>: Invalid message received from %s: header mismatch", appName)
		return
	case len(msg[3]) != 1:
		log.Warnf("zmq3<PubSub>: Invalid message received from %v: invalid message type", appName)
		return
	}
	switch msg[3][0] {
	case MessageTypeEvent:
		if len(msg) != 6 || len(msg[1]) == 0 || len(msg[4]) != 0 {
			log.Warnf("zmq3<PubSub>: Invalid EVENT message received from %v", appName)
			return
		}
		log.Debugf("zmq3<PubSub>: EVENT message received from %v", appName)
		ep.exchange.Publish(Event(msg))
	case MessageTypeSeqTable:
		if len(msg) != 4 || len(msg[1]) == 0 {
			log.Warnf("zmq3<PubSub>: Invalid SEQTABLE message received from %v", appName)
			return
		}
		log.Debugf("zmq3<PubSub>: SEQTABLE message received from %v", appName)
		table := ep.exchange.EventSequenceTable(msg[1])
		ep.sendEventSequenceTable(msg[0], table)
	default:
		log.Warnf("zmq3<PubSub>: Unknown message type received from %v", appName)
	}
}
Processing of incoming messages  --------------------------------------------
package pubsub
import (
	"bytes"
	"encoding/binary"
	"errors"
	"unsafe"
	"github.com/meeko/meekod/broker/log"
	"github.com/meeko/meekod/broker/services/pubsub"
	"github.com/meeko/meekod/broker/transports/zmq3/loop"
	"github.com/dmotylev/nutrition"
	zmq "github.com/pebbe/zmq3"
)
const Header = "CDR#PUBSUB@01"
const (
	MessageTypeEvent byte = iota
	MessageTypeSeqTable
)
var (
	FrameHeader = []byte(Header)
	FrameEventType    = []byte{MessageTypeEvent}
	FrameSeqTableType = []byte{MessageTypeSeqTable}
)
var (
	ErrConfigIncomplete = errors.New("PubSub endpoint is not fully configured")
	ErrSeqNumNotSet     = errors.New("Event sequence number is not set")
)
type EndpointConfig struct {
	RouterEndpoint string
	RouterSndhwm   int
	RouterRcvhwm   int
	PubSndhwm      int
	PubEndpoint    string
}
func NewEndpointConfig() *EndpointConfig {
	return &EndpointConfig{
		RouterSndhwm: 1000,
		RouterRcvhwm: 1000,
		PubSndhwm:    1000,
	}
}
func (config *EndpointConfig) FeedFromEnv(prefix string) error {
	log.Debug("zmq3<PubSub>: Feeding config from the environment")
	return nutrition.Env(prefix).Feed(config)
}
func (config *EndpointConfig) MustFeedFromEnv(prefix string) *EndpointConfig {
	if err := config.FeedFromEnv(prefix); err != nil {
		panic(err)
	}
	return config
}
func (config *EndpointConfig) IsComplete() bool {
	return config.PubEndpoint != "" && config.RouterEndpoint != ""
}
func (config *EndpointConfig) MustBeComplete() {
	if !config.IsComplete() {
		panic(ErrConfigIncomplete)
	}
}
type Endpoint struct {
	*loop.MessageLoop
	exchange pubsub.Exchange
}
func NewEndpoint(config *EndpointConfig, exchange pubsub.Exchange) (pubsub.Endpoint, error) {
	if !config.IsComplete() {
		return nil, ErrConfigIncomplete
	}
	log.Debugf("zmq3<PubSub>: Instantiating endpoint using %#v", config)
	router, err := zmq.NewSocket(zmq.ROUTER)
	if err != nil {
		return nil, err
	}
	if err := router.SetSndhwm(config.RouterSndhwm); err != nil {
		router.Close()
		return nil, err
	}
	if err := router.SetRcvhwm(config.RouterRcvhwm); err != nil {
		router.Close()
		return nil, err
	}
	if err := router.Bind(config.RouterEndpoint); err != nil {
		router.Close()
		return nil, err
	}
	pub, err := zmq.NewSocket(zmq.PUB)
	if err != nil {
		router.Close()
		return nil, err
	}
	if err := pub.SetSndhwm(config.PubSndhwm); err != nil {
		router.Close()
		return nil, err
	}
	if err = pub.Bind(config.PubEndpoint); err != nil {
		router.Close()
		return nil, err
	}
	ep := &Endpoint{
		exchange: exchange,
	}
	messageLoop := loop.New(loop.PollItems{
		{pub, nil},
		{router, ep.handleMessage},
	})
	exchange.RegisterEndpoint(ep)
	ep.MessageLoop = messageLoop
	return ep, nil
}
func (ep *Endpoint) Publish(event pubsub.Event) error {
	if len(event.Seq()) != 4 {  
		return ErrSeqNumNotSet
	}
	e, ok := event.(Event)
	if ok {
		msg := [][]byte(e)
		tmp := msg[0]
		msg[0] = msg[1]
		msg[1] = tmp
		err := ep.MessageLoop.DispatchMessage(0, msg)
		tmp = msg[0]
		msg[0] = msg[1]
		msg[1] = tmp
		return err
	}
	var seqBuf bytes.Buffer
	if err := binary.Write(&seqBuf, binary.BigEndian, event.Seq()); err != nil {
		return err
	}
	return ep.MessageLoop.DispatchMessage(0, [][]byte{
		event.Kind(),
		event.Publisher(),
		FrameHeader,
		FrameEventType,
		seqBuf.Bytes(),
		event.Body(),
	})
}
func (ep *Endpoint) Close() error {
	ep.exchange.UnregisterEndpoint(ep)
	return ep.MessageLoop.Close()
}