func (h *Hookbot) Loop() {
	defer h.wg.Done()
	cMessageListener := make(chan MessageListener, 1000)
	defer close(cMessageListener)
	var wgMessageListeners sync.WaitGroup
	defer wgMessageListeners.Wait()
	cMessageListeners := make(chan MessageListeners, 100)
	defer close(cMessageListeners)
	const W = 10  
	for i := 0; i < W; i++ {
		h.wg.Add(1)
		go func() {
			defer h.wg.Done()
			h.TimeoutSendWorker(cMessageListener)
		}()
	}
	wgMessageListeners.Add(1)
	go func() {
		defer wgMessageListeners.Done()
		for lms := range cMessageListeners {
			for l := range lms.interested {
				cMessageListener <- MessageListener{l, lms.m}
			}
		}
	}()
	listeners := map[string]map[Listener]struct{}{}
	interested := func(topic string) map[Listener]struct{} {
		ls := map[Listener]struct{}{}
		for l := range listeners[topic] {
			ls[l] = struct{}{}
		}
		for fullCandidateTopic, candidateLs := range listeners {
			candidateTopic, isRec := recursive(fullCandidateTopic)
			if !isRec {
				continue
			}
			if !strings.HasPrefix(topic, candidateTopic) {
				continue
			}
			for l := range candidateLs {
				ls[l] = struct{}{}
			}
		}
		return ls
	}
	for {
		select {
		case m := <-h.message:
			select {
			case cMessageListeners <- MessageListeners{interested(m.Topic), &m}:
				atomic.AddInt64(&h.publish, 1)
				m.Sent <- true
			default:
				atomic.AddInt64(&h.dropP, 1)
				m.Sent <- false
			}
		case l := <-h.addListener:
			atomic.AddInt64(&h.listeners, 1)
			if _, ok := listeners[l.Topic]; !ok {
				listeners[l.Topic] = map[Listener]struct{}{}
			}
			listeners[l.Topic][l] = struct{}{}
			close(l.ready)
		case l := <-h.delListener:
			atomic.AddInt64(&h.listeners, -1)
			delete(listeners[l.Topic], l)
			if len(listeners[l.Topic]) == 0 {
				delete(listeners, l.Topic)
			}
		case <-h.shutdown:
			return
		}
	}
}
Manage fanout from h.message onto listeners
package hookbot
import (
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"net/http"
	"regexp"
	"strings"
	"sync"
	"sync/atomic"
	"time"
	"github.com/gorilla/websocket"
)
type Message struct {
	Topic string
	Body  []byte
	Sent chan bool  
}
type Listener struct {
	Topic string
	c     chan Message
	ready chan struct{}  
	dead  chan struct{}  
}
type Hookbot struct {
	key string
	wg       *sync.WaitGroup
	shutdown chan struct{}
	http.Handler
	message                  chan Message
	addListener, delListener chan Listener
	routers []Router
	listeners, publish, dropP, sends, dropS int64
}
func New(key string) *Hookbot {
	h := &Hookbot{
		key: key,
		wg:       &sync.WaitGroup{},
		shutdown: make(chan struct{}),
		message:     make(chan Message, 1),
		addListener: make(chan Listener, 1),
		delListener: make(chan Listener, 1),
	}
	sub := WebsocketHandlerFunc(h.ServeSubscribe)
	pub := http.HandlerFunc(h.ServePublish)
	mux := http.NewServeMux()
	mux.Handle("/sub/", h.KeyChecker(sub))
	mux.Handle("/pub/", h.KeyChecker(pub))
	mux.Handle("/unsafe/sub/", RequireUnsafeHeader(h.KeyChecker(sub)))
	mux.Handle("/unsafe/pub/", pub)
	mux.Handle("/", h.KeyChecker(h.BothPubSub(pub, sub)))
	h.Handler = mux
	h.wg.Add(1)
	go h.Loop()
	h.wg.Add(1)
	go h.ShowStatus(time.Minute)
	return h
}
func (h *Hookbot) BothPubSub(pub, sub http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		switch r.Method {
		case "POST":
			pub.ServeHTTP(w, r)
		case "GET":
			sub.ServeHTTP(w, r)
		default:
			http.Error(w, "Not Implemented", http.StatusNotImplemented)
		}
	})
}
func (h *Hookbot) ShowStatus(period time.Duration) {
	defer h.wg.Done()
	ticker := time.NewTicker(period)
	var ll, lp, ls, ldP, ldS int64
	for {
		select {
		case <-ticker.C:
			l := atomic.LoadInt64(&h.listeners)
			p := atomic.LoadInt64(&h.publish)
			s := atomic.LoadInt64(&h.sends)
			dP := atomic.LoadInt64(&h.dropP)
			dS := atomic.LoadInt64(&h.dropS)
			log.Printf("Listeners %5d [%+5d] pub %5d [%+5d] (d %5d [%+5d])"+
				" send %8d [%+7d] (d %5d [%+5d])",
				l, l-ll, p, p-lp, dP, dP-ldP, s, s-ls, dS, dS-ldS)
			ll, lp, ls, ldP, ldS = l, p, s, dP, dS
		case <-h.shutdown:
			return
		}
	}
}
func (h *Hookbot) Shutdown() {
	close(h.shutdown)
	h.wg.Wait()
}
func recursive(fullTopic string) (topic string, isRecursive bool) {
	if strings.HasSuffix(fullTopic, "?recursive") {
		return fullTopic[:len(fullTopic)-len("?recursive")], true
	}
	if strings.HasSuffix(fullTopic, "/") {
		return fullTopic, true
	}
	return fullTopic, false
}
type MessageListener struct {
	l Listener
	m *Message
}
type MessageListeners struct {
	interested map[Listener]struct{}
	m          *Message
}
const timeout = 1 * time.Second
func (h *Hookbot) TimeoutSendWorker(r chan MessageListener) {
	for lm := range r {
		select {
		case lm.l.c <- *lm.m:
			atomic.AddInt64(&h.sends, 1)
		case <-time.After(timeout):
			atomic.AddInt64(&h.dropS, 1)
		case <-lm.l.dead:
		}
	}
}