2017-06-03 15:16:17 +00:00
|
|
|
package msgbus
|
|
|
|
|
|
|
|
import (
|
2022-03-29 22:03:38 +00:00
|
|
|
"compress/flate"
|
|
|
|
"compress/gzip"
|
2017-08-09 09:54:11 +00:00
|
|
|
"encoding/json"
|
2022-04-04 01:15:49 +00:00
|
|
|
"errors"
|
2018-04-06 08:27:25 +00:00
|
|
|
"fmt"
|
2022-03-29 22:03:38 +00:00
|
|
|
"io"
|
2022-04-03 15:59:38 +00:00
|
|
|
"io/fs"
|
2017-08-09 09:54:11 +00:00
|
|
|
"io/ioutil"
|
|
|
|
"net/http"
|
2022-04-03 15:59:38 +00:00
|
|
|
"os"
|
|
|
|
"path/filepath"
|
2022-03-27 02:42:56 +00:00
|
|
|
"strconv"
|
2017-08-09 09:54:11 +00:00
|
|
|
"strings"
|
2017-06-03 15:16:17 +00:00
|
|
|
"time"
|
2017-08-09 09:54:11 +00:00
|
|
|
|
2022-03-29 22:03:38 +00:00
|
|
|
"github.com/andybalholm/brotli"
|
2022-04-03 15:59:38 +00:00
|
|
|
securejoin "github.com/cyphar/filepath-securejoin"
|
2022-04-02 14:05:15 +00:00
|
|
|
sync "github.com/sasha-s/go-deadlock"
|
2018-03-26 00:03:56 +00:00
|
|
|
log "github.com/sirupsen/logrus"
|
2022-04-03 15:59:38 +00:00
|
|
|
"github.com/tidwall/wal"
|
|
|
|
msgpack "github.com/vmihailenco/msgpack/v5"
|
2018-03-26 00:03:56 +00:00
|
|
|
|
2022-03-27 02:42:56 +00:00
|
|
|
"github.com/gorilla/websocket"
|
2017-06-03 15:16:17 +00:00
|
|
|
)
|
|
|
|
|
2017-08-14 07:34:12 +00:00
|
|
|
const (
|
2018-05-03 07:57:52 +00:00
|
|
|
// Time allowed to write a message to the peer.
|
|
|
|
writeWait = 10 * time.Second
|
|
|
|
|
|
|
|
// Time allowed to read the next pong message from the peer.
|
|
|
|
pongWait = 60 * time.Second
|
|
|
|
|
|
|
|
// Send pings to peer with this period. Must be less than pongWait.
|
|
|
|
pingPeriod = (pongWait * 9) / 10
|
2017-08-14 07:34:12 +00:00
|
|
|
)
|
|
|
|
|
2022-04-04 01:15:49 +00:00
|
|
|
var (
|
|
|
|
// BufferFull is logged in Subscribe() when a subscriber's
|
|
|
|
// buffer is full and messages can no longer be enqueued for delivery
|
|
|
|
ErrBufferFull = errors.New("error: subscriber buffer full")
|
|
|
|
)
|
|
|
|
|
2022-03-27 02:42:56 +00:00
|
|
|
// TODO: Make this configurable?
|
|
|
|
var upgrader = websocket.Upgrader{
|
|
|
|
ReadBufferSize: 4096,
|
|
|
|
WriteBufferSize: 4096,
|
|
|
|
CheckOrigin: func(r *http.Request) bool {
|
|
|
|
return true
|
|
|
|
},
|
2018-05-02 06:19:04 +00:00
|
|
|
}
|
|
|
|
|
2018-05-01 06:09:12 +00:00
|
|
|
// HandlerFunc ...
|
|
|
|
type HandlerFunc func(msg *Message) error
|
|
|
|
|
2017-08-14 07:34:12 +00:00
|
|
|
// Topic ...
|
|
|
|
type Topic struct {
|
2018-05-10 06:25:13 +00:00
|
|
|
Name string `json:"name"`
|
2022-04-03 15:59:38 +00:00
|
|
|
Sequence int64 `json:"seq"`
|
2018-05-10 06:25:13 +00:00
|
|
|
Created time.Time `json:"created"`
|
2017-08-14 07:34:12 +00:00
|
|
|
}
|
|
|
|
|
2018-12-31 08:05:23 +00:00
|
|
|
func (t *Topic) String() string {
|
|
|
|
return t.Name
|
|
|
|
}
|
|
|
|
|
2017-06-03 15:16:17 +00:00
|
|
|
// Message ...
|
|
|
|
type Message struct {
|
2022-04-03 15:59:38 +00:00
|
|
|
ID int64 `json:"id"`
|
2017-08-14 07:34:12 +00:00
|
|
|
Topic *Topic `json:"topic"`
|
2017-06-03 15:16:17 +00:00
|
|
|
Payload []byte `json:"payload"`
|
|
|
|
Created time.Time `json:"created"`
|
|
|
|
}
|
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
func LoadMessage(data []byte) (m Message, err error) {
|
|
|
|
err = msgpack.Unmarshal(data, &m)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
func (m Message) Bytes() ([]byte, error) {
|
|
|
|
return msgpack.Marshal(m)
|
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
// SubscribeOption ...
|
|
|
|
type SubscribeOption func(*SubscriberOptions)
|
|
|
|
|
|
|
|
// SubscriberOptions ...
|
|
|
|
type SubscriberOptions struct {
|
2022-04-03 15:59:38 +00:00
|
|
|
Index int64
|
2022-04-02 14:05:15 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// WithIndex sets the index to start subscribing from
|
2022-04-03 15:59:38 +00:00
|
|
|
func WithIndex(index int64) SubscribeOption {
|
2022-04-02 14:05:15 +00:00
|
|
|
return func(o *SubscriberOptions) { o.Index = index }
|
|
|
|
}
|
|
|
|
|
|
|
|
// SubscribersConfig ...
|
|
|
|
type SubscriberConfig struct {
|
2018-05-14 09:12:58 +00:00
|
|
|
BufferLength int
|
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
// Subscribers ...
|
|
|
|
type Subscribers struct {
|
2018-05-10 09:46:25 +00:00
|
|
|
sync.RWMutex
|
|
|
|
|
2018-05-14 09:12:58 +00:00
|
|
|
buflen int
|
|
|
|
|
2017-06-03 15:16:17 +00:00
|
|
|
ids map[string]bool
|
2017-08-07 07:31:53 +00:00
|
|
|
chs map[string]chan Message
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
// NewSubscribers ...
|
|
|
|
func NewSubscribers(config *SubscriberConfig) *Subscribers {
|
2018-05-14 09:12:58 +00:00
|
|
|
var (
|
|
|
|
bufferLength int
|
|
|
|
)
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
if config != nil {
|
|
|
|
bufferLength = config.BufferLength
|
2018-05-14 09:12:58 +00:00
|
|
|
} else {
|
|
|
|
bufferLength = DefaultBufferLength
|
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
return &Subscribers{
|
2018-05-14 09:12:58 +00:00
|
|
|
buflen: bufferLength,
|
|
|
|
|
2017-06-03 15:16:17 +00:00
|
|
|
ids: make(map[string]bool),
|
2017-08-07 07:31:53 +00:00
|
|
|
chs: make(map[string]chan Message),
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
// Len ...
|
|
|
|
func (subs *Subscribers) Len() int {
|
|
|
|
subs.RLock()
|
|
|
|
defer subs.RUnlock()
|
2018-05-10 09:46:25 +00:00
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
return len(subs.ids)
|
2018-05-05 08:29:08 +00:00
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
// AddSubscriber ...
|
|
|
|
func (subs *Subscribers) AddSubscriber(id string) chan Message {
|
|
|
|
subs.Lock()
|
|
|
|
defer subs.Unlock()
|
|
|
|
|
|
|
|
if subs.buflen <= 0 {
|
|
|
|
log.Fatal("subscriber buflen <= 0")
|
|
|
|
}
|
|
|
|
|
|
|
|
ch := make(chan Message, subs.buflen)
|
|
|
|
|
|
|
|
subs.ids[id] = true
|
|
|
|
subs.chs[id] = ch
|
2018-05-10 09:46:25 +00:00
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
return ch
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
// RemoveSubscriber ...
|
|
|
|
func (subs *Subscribers) RemoveSubscriber(id string) {
|
|
|
|
subs.Lock()
|
|
|
|
defer subs.Unlock()
|
2018-05-10 09:46:25 +00:00
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
delete(subs.ids, id)
|
2017-06-03 15:16:17 +00:00
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
close(subs.chs[id])
|
|
|
|
delete(subs.chs, id)
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
// HasSubscriber ...
|
|
|
|
func (subs *Subscribers) HasSubscriber(id string) bool {
|
|
|
|
subs.RLock()
|
|
|
|
defer subs.RUnlock()
|
2018-05-10 09:46:25 +00:00
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
_, ok := subs.ids[id]
|
2017-06-03 15:16:17 +00:00
|
|
|
return ok
|
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
// GetSubscriber ...
|
|
|
|
func (subs *Subscribers) GetSubscriber(id string) (chan Message, bool) {
|
|
|
|
subs.RLock()
|
|
|
|
defer subs.RUnlock()
|
2018-05-10 09:46:25 +00:00
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
ch, ok := subs.chs[id]
|
2017-06-03 15:16:17 +00:00
|
|
|
if !ok {
|
|
|
|
return nil, false
|
|
|
|
}
|
|
|
|
return ch, true
|
|
|
|
}
|
|
|
|
|
|
|
|
// NotifyAll ...
|
2022-04-02 14:05:15 +00:00
|
|
|
func (subs *Subscribers) NotifyAll(message Message) int {
|
|
|
|
subs.RLock()
|
|
|
|
defer subs.RUnlock()
|
2018-05-10 09:46:25 +00:00
|
|
|
|
2018-05-05 08:29:08 +00:00
|
|
|
i := 0
|
2022-04-02 14:05:15 +00:00
|
|
|
for id, ch := range subs.chs {
|
2018-05-03 18:04:15 +00:00
|
|
|
select {
|
|
|
|
case ch <- message:
|
2018-05-05 08:29:08 +00:00
|
|
|
i++
|
2018-05-03 18:04:15 +00:00
|
|
|
default:
|
|
|
|
// TODO: Drop this client?
|
|
|
|
// TODO: Retry later?
|
2022-04-04 01:15:49 +00:00
|
|
|
log.Warnf("cannot publish message %s#%d to %s", message.Topic.Name, message.ID, id)
|
2018-05-03 18:04:15 +00:00
|
|
|
}
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
2018-05-05 08:29:08 +00:00
|
|
|
|
|
|
|
return i
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// MessageBus ...
|
|
|
|
type MessageBus struct {
|
2018-05-10 09:46:25 +00:00
|
|
|
sync.RWMutex
|
2018-04-02 21:38:33 +00:00
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
options *Options
|
2018-05-02 07:41:14 +00:00
|
|
|
metrics *Metrics
|
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
topics map[string]*Topic
|
|
|
|
queues map[*Topic]*Queue
|
|
|
|
logs map[*Topic]*wal.Log
|
2018-05-10 06:25:13 +00:00
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
subscribers map[*Topic]*Subscribers
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
// NewMessageBus creates a new message bus with the provided options
|
|
|
|
func NewMessageBus(opts ...Option) (*MessageBus, error) {
|
2022-04-04 01:15:49 +00:00
|
|
|
options := NewDefaultOptions()
|
2017-08-14 07:34:12 +00:00
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
for _, opt := range opts {
|
|
|
|
if err := opt(options); err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2018-05-02 08:24:30 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
var metrics *Metrics
|
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
if options.Metrics {
|
2018-05-02 08:24:30 +00:00
|
|
|
metrics = NewMetrics("msgbus")
|
|
|
|
|
|
|
|
ctime := time.Now()
|
|
|
|
|
|
|
|
// server uptime counter
|
|
|
|
metrics.NewCounterFunc(
|
|
|
|
"server", "uptime",
|
|
|
|
"Number of nanoseconds the server has been running",
|
|
|
|
func() float64 {
|
|
|
|
return float64(time.Since(ctime).Nanoseconds())
|
|
|
|
},
|
|
|
|
)
|
|
|
|
|
|
|
|
// server requests counter
|
|
|
|
metrics.NewCounter(
|
2018-05-05 08:29:08 +00:00
|
|
|
"server", "requests",
|
|
|
|
"Number of total requests processed",
|
|
|
|
)
|
|
|
|
|
|
|
|
// client latency summary
|
|
|
|
metrics.NewSummary(
|
|
|
|
"client", "latency_seconds",
|
|
|
|
"Client latency in seconds",
|
|
|
|
)
|
|
|
|
|
|
|
|
// client errors counter
|
|
|
|
metrics.NewCounter(
|
|
|
|
"client", "errors",
|
|
|
|
"Number of errors publishing messages to clients",
|
2018-05-02 08:24:30 +00:00
|
|
|
)
|
|
|
|
|
|
|
|
// bus messages counter
|
|
|
|
metrics.NewCounter(
|
|
|
|
"bus", "messages",
|
|
|
|
"Number of total messages exchanged",
|
|
|
|
)
|
|
|
|
|
2018-05-02 19:50:12 +00:00
|
|
|
// bus dropped counter
|
|
|
|
metrics.NewCounter(
|
|
|
|
"bus", "dropped",
|
|
|
|
"Number of messages dropped to subscribers",
|
|
|
|
)
|
|
|
|
|
|
|
|
// bus delivered counter
|
|
|
|
metrics.NewCounter(
|
|
|
|
"bus", "delivered",
|
|
|
|
"Number of messages delivered to subscribers",
|
|
|
|
)
|
|
|
|
|
|
|
|
// bus fetched counter
|
|
|
|
metrics.NewCounter(
|
|
|
|
"bus", "fetched",
|
|
|
|
"Number of messages fetched from clients",
|
|
|
|
)
|
|
|
|
|
2018-05-02 08:24:30 +00:00
|
|
|
// bus topics gauge
|
|
|
|
metrics.NewCounter(
|
|
|
|
"bus", "topics",
|
|
|
|
"Number of active topics registered",
|
|
|
|
)
|
|
|
|
|
2018-05-10 06:25:13 +00:00
|
|
|
// queue len gauge vec
|
|
|
|
metrics.NewGaugeVec(
|
|
|
|
"queue", "len",
|
|
|
|
"Queue length of each topic",
|
|
|
|
[]string{"topic"},
|
|
|
|
)
|
|
|
|
|
|
|
|
// queue size gauge vec
|
|
|
|
// TODO: Implement this gauge by somehow getting queue sizes per topic!
|
2018-05-08 04:51:54 +00:00
|
|
|
metrics.NewGaugeVec(
|
2018-05-10 06:25:13 +00:00
|
|
|
"queue", "size",
|
|
|
|
"Queue length of each topic",
|
2018-05-08 04:51:54 +00:00
|
|
|
[]string{"topic"},
|
|
|
|
)
|
|
|
|
|
2018-05-02 08:24:30 +00:00
|
|
|
// bus subscribers gauge
|
|
|
|
metrics.NewGauge(
|
|
|
|
"bus", "subscribers",
|
|
|
|
"Number of active subscribers",
|
|
|
|
)
|
2017-08-14 07:34:12 +00:00
|
|
|
}
|
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
mb := &MessageBus{
|
|
|
|
options: options,
|
2018-05-02 07:41:14 +00:00
|
|
|
metrics: metrics,
|
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
topics: make(map[string]*Topic),
|
|
|
|
queues: make(map[*Topic]*Queue),
|
|
|
|
logs: make(map[*Topic]*wal.Log),
|
2018-05-10 06:25:13 +00:00
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
subscribers: make(map[*Topic]*Subscribers),
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
2022-04-03 15:59:38 +00:00
|
|
|
|
|
|
|
if err := mb.readLogs(); err != nil {
|
|
|
|
return nil, fmt.Errorf("error reading logs: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
return mb, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (mb *MessageBus) readLog(f fs.FileInfo) error {
|
|
|
|
log.Debugf("reading log %s", f.Name())
|
|
|
|
|
|
|
|
l, err := wal.Open(filepath.Join(mb.options.LogPath, f.Name()), nil)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error opening log %s: %w", f, err)
|
|
|
|
}
|
|
|
|
|
|
|
|
t := mb.newTopic(f.Name())
|
|
|
|
t.Created = f.ModTime()
|
|
|
|
|
|
|
|
first, err := l.FirstIndex()
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error reading first index: %w", err)
|
|
|
|
}
|
|
|
|
last, err := l.LastIndex()
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error reading last index: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
log.Debugf("first index: %d", first)
|
|
|
|
log.Debugf("last index: %d", last)
|
|
|
|
|
|
|
|
t.Sequence = int64(last)
|
|
|
|
|
|
|
|
q := NewQueue(mb.options.MaxQueueSize)
|
|
|
|
|
|
|
|
start := int64(last) - int64(mb.options.MaxQueueSize)
|
|
|
|
if start < 0 {
|
|
|
|
start = int64(first)
|
|
|
|
}
|
|
|
|
end := int64(last)
|
|
|
|
log.Debugf("start: %d", start)
|
|
|
|
log.Debugf("end: %d", end)
|
|
|
|
for i := start; i <= end; i++ {
|
|
|
|
data, err := l.Read(uint64(i))
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error reading log %d: %w", i, err)
|
|
|
|
}
|
|
|
|
|
|
|
|
msg, err := LoadMessage(data)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error deserialing log %d: %w", i, err)
|
|
|
|
}
|
|
|
|
|
|
|
|
q.Push(msg)
|
|
|
|
}
|
|
|
|
mb.queues[t] = q
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (mb *MessageBus) readLogs() error {
|
|
|
|
mb.Lock()
|
|
|
|
defer mb.Unlock()
|
|
|
|
|
|
|
|
log.Debug("reading logs...")
|
|
|
|
|
|
|
|
dirs, err := os.ReadDir(mb.options.LogPath)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error listing logs: %w", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, dir := range dirs {
|
|
|
|
if dir.IsDir() {
|
|
|
|
info, err := dir.Info()
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error reading log path %s: %w", dir, err)
|
|
|
|
}
|
|
|
|
if err := mb.readLog(info); err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// Len ...
|
|
|
|
func (mb *MessageBus) Len() int {
|
|
|
|
return len(mb.topics)
|
|
|
|
}
|
|
|
|
|
2018-05-02 08:24:30 +00:00
|
|
|
// Metrics ...
|
|
|
|
func (mb *MessageBus) Metrics() *Metrics {
|
|
|
|
return mb.metrics
|
|
|
|
}
|
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
func (mb *MessageBus) newTopic(topic string) *Topic {
|
2017-08-14 07:34:12 +00:00
|
|
|
t, ok := mb.topics[topic]
|
|
|
|
if !ok {
|
2018-05-10 06:25:13 +00:00
|
|
|
t = &Topic{Name: topic, Created: time.Now()}
|
2017-08-14 07:34:12 +00:00
|
|
|
mb.topics[topic] = t
|
2018-05-02 08:24:30 +00:00
|
|
|
if mb.metrics != nil {
|
|
|
|
mb.metrics.Counter("bus", "topics").Inc()
|
|
|
|
}
|
2017-08-14 07:34:12 +00:00
|
|
|
}
|
|
|
|
return t
|
|
|
|
}
|
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
// NewTopic ...
|
|
|
|
func (mb *MessageBus) NewTopic(topic string) *Topic {
|
|
|
|
mb.Lock()
|
|
|
|
defer mb.Unlock()
|
|
|
|
|
|
|
|
return mb.newTopic(topic)
|
|
|
|
}
|
|
|
|
|
2017-06-03 15:16:17 +00:00
|
|
|
// NewMessage ...
|
2017-08-14 07:34:12 +00:00
|
|
|
func (mb *MessageBus) NewMessage(topic *Topic, payload []byte) Message {
|
|
|
|
defer func() {
|
|
|
|
topic.Sequence++
|
2018-05-02 08:24:30 +00:00
|
|
|
if mb.metrics != nil {
|
|
|
|
mb.metrics.Counter("bus", "messages").Inc()
|
|
|
|
}
|
2017-08-14 07:34:12 +00:00
|
|
|
}()
|
|
|
|
|
|
|
|
return Message{
|
2022-04-03 15:59:38 +00:00
|
|
|
ID: topic.Sequence + 1,
|
2017-08-14 07:34:12 +00:00
|
|
|
Topic: topic,
|
2017-06-03 15:16:17 +00:00
|
|
|
Payload: payload,
|
|
|
|
Created: time.Now(),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Put ...
|
2022-04-03 15:59:38 +00:00
|
|
|
func (mb *MessageBus) Put(message Message) error {
|
2018-05-10 09:46:25 +00:00
|
|
|
mb.Lock()
|
|
|
|
defer mb.Unlock()
|
|
|
|
|
2018-05-08 04:51:54 +00:00
|
|
|
t := message.Topic
|
2022-04-03 15:59:38 +00:00
|
|
|
|
|
|
|
l, ok := mb.logs[t]
|
|
|
|
if !ok {
|
|
|
|
fn, err := securejoin.SecureJoin(mb.options.LogPath, t.Name)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error creating logfile filename for %s: %w", t.Name, err)
|
|
|
|
}
|
|
|
|
|
|
|
|
l, err = wal.Open(fn, nil)
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error opening logfile %s: %w", fn, err)
|
|
|
|
}
|
|
|
|
mb.logs[t] = l
|
|
|
|
}
|
|
|
|
|
|
|
|
id := uint64(message.ID)
|
|
|
|
|
|
|
|
data, err := message.Bytes()
|
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("error serializing message %d: %w", id, err)
|
|
|
|
}
|
|
|
|
|
|
|
|
if err := l.Write(id, data); err != nil {
|
|
|
|
return fmt.Errorf("error writing message %d to logfile: %w", message.ID, err)
|
|
|
|
}
|
|
|
|
|
2018-05-08 04:51:54 +00:00
|
|
|
q, ok := mb.queues[t]
|
2017-06-03 15:16:17 +00:00
|
|
|
if !ok {
|
2022-04-03 15:59:38 +00:00
|
|
|
q = NewQueue(mb.options.MaxQueueSize)
|
|
|
|
mb.queues[t] = q
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
q.Push(message)
|
|
|
|
|
2018-05-08 04:51:54 +00:00
|
|
|
if mb.metrics != nil {
|
2018-05-10 06:25:13 +00:00
|
|
|
mb.metrics.GaugeVec("queue", "len").WithLabelValues(t.Name).Inc()
|
2018-05-08 04:51:54 +00:00
|
|
|
}
|
|
|
|
|
2018-05-10 09:46:25 +00:00
|
|
|
mb.publish(message)
|
2022-04-03 15:59:38 +00:00
|
|
|
|
|
|
|
return nil
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// Get ...
|
2018-05-08 04:51:54 +00:00
|
|
|
func (mb *MessageBus) Get(t *Topic) (Message, bool) {
|
2018-05-10 09:46:25 +00:00
|
|
|
mb.RLock()
|
|
|
|
defer mb.RUnlock()
|
|
|
|
|
2018-05-08 04:51:54 +00:00
|
|
|
q, ok := mb.queues[t]
|
2017-06-03 15:16:17 +00:00
|
|
|
if !ok {
|
2017-08-07 07:31:53 +00:00
|
|
|
return Message{}, false
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
m := q.Pop()
|
|
|
|
if m == nil {
|
2017-08-07 07:31:53 +00:00
|
|
|
return Message{}, false
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
2018-05-02 19:50:12 +00:00
|
|
|
|
|
|
|
if mb.metrics != nil {
|
|
|
|
mb.metrics.Counter("bus", "fetched").Inc()
|
2018-05-10 06:25:13 +00:00
|
|
|
mb.metrics.GaugeVec("queue", "len").WithLabelValues(t.Name).Dec()
|
2018-05-02 19:50:12 +00:00
|
|
|
}
|
|
|
|
|
2017-08-07 07:31:53 +00:00
|
|
|
return m.(Message), true
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
2018-05-10 09:46:25 +00:00
|
|
|
// publish ...
|
|
|
|
func (mb *MessageBus) publish(message Message) {
|
2022-04-02 14:05:15 +00:00
|
|
|
subs, ok := mb.subscribers[message.Topic]
|
2017-06-03 15:16:17 +00:00
|
|
|
if !ok {
|
2022-04-02 14:05:15 +00:00
|
|
|
log.Debugf("no subscribers for %s", message.Topic.Name)
|
2017-06-03 15:16:17 +00:00
|
|
|
return
|
|
|
|
}
|
2018-05-05 08:29:08 +00:00
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
log.Debug("notifying subscribers")
|
|
|
|
n := subs.NotifyAll(message)
|
|
|
|
if n != subs.Len() && mb.metrics != nil {
|
|
|
|
log.Warnf("%d/%d subscribers notified", n, subs.Len())
|
|
|
|
mb.metrics.Counter("bus", "dropped").Add(float64(subs.Len() - n))
|
2018-05-05 08:29:08 +00:00
|
|
|
}
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// Subscribe ...
|
2022-04-02 14:05:15 +00:00
|
|
|
func (mb *MessageBus) Subscribe(id, topic string, opts ...SubscribeOption) chan Message {
|
2018-05-10 09:46:25 +00:00
|
|
|
mb.Lock()
|
|
|
|
defer mb.Unlock()
|
|
|
|
|
2017-08-14 07:34:12 +00:00
|
|
|
t, ok := mb.topics[topic]
|
|
|
|
if !ok {
|
2018-05-10 06:25:13 +00:00
|
|
|
t = &Topic{Name: topic, Created: time.Now()}
|
2017-08-14 07:34:12 +00:00
|
|
|
mb.topics[topic] = t
|
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
subs, ok := mb.subscribers[t]
|
2017-06-03 15:16:17 +00:00
|
|
|
if !ok {
|
2022-04-03 15:59:38 +00:00
|
|
|
subs = NewSubscribers(&SubscriberConfig{BufferLength: mb.options.BufferLength})
|
2022-04-02 14:05:15 +00:00
|
|
|
mb.subscribers[t] = subs
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
if subs.HasSubscriber(id) {
|
2018-05-08 07:18:34 +00:00
|
|
|
// Already verified the listener exists
|
2022-04-02 14:05:15 +00:00
|
|
|
log.Debugf("already have subscriber %s", id)
|
|
|
|
ch, _ := subs.GetSubscriber(id)
|
2017-06-03 15:16:17 +00:00
|
|
|
return ch
|
|
|
|
}
|
2018-05-08 07:18:34 +00:00
|
|
|
|
|
|
|
if mb.metrics != nil {
|
|
|
|
mb.metrics.Gauge("bus", "subscribers").Inc()
|
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
o := &SubscriberOptions{}
|
|
|
|
for _, opt := range opts {
|
|
|
|
opt(o)
|
|
|
|
}
|
|
|
|
|
|
|
|
ch := subs.AddSubscriber(id)
|
|
|
|
q, ok := mb.queues[t]
|
|
|
|
if !ok {
|
|
|
|
log.Debug("nothing in queue, returning ch")
|
|
|
|
return ch
|
|
|
|
}
|
|
|
|
|
2022-04-02 22:23:01 +00:00
|
|
|
// IF client requests to start from index >= 0 (-1 from the head)
|
|
|
|
// AND the topic's sequence number hasn't been reset (< Index) THEN
|
2022-04-03 15:59:38 +00:00
|
|
|
if o.Index > 0 && o.Index <= t.Sequence {
|
2022-04-02 14:05:15 +00:00
|
|
|
var n int
|
|
|
|
log.Debugf("subscriber wants to start from %d", o.Index)
|
2022-04-04 01:15:49 +00:00
|
|
|
err := q.ForEach(func(item interface{}) error {
|
Fix panic on nil message (#29)
Fixed a bug I found in `saltyd`:
```
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | 2022/04/02 21:49:31 http: panic serving 10.0.2.82:50050: interface conversion: interface {} is nil, not msgbus.Message
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | goroutine 1839 [running]:
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | net/http.(*conn).serve.func1()
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /usr/local/go/src/net/http/server.go:1825 +0xbf
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | panic({0xc8c2c0, 0xc000095e60})
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /usr/local/go/src/runtime/panic.go:844 +0x258
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | git.mills.io/prologic/msgbus.(*MessageBus).Subscribe.func1({0x0?, 0x0?})
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /go/pkg/mod/git.mills.io/prologic/msgbus@v0.1.13/msgbus.go:487 +0x205
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | git.mills.io/prologic/msgbus.(*Queue).ForEach(0xc00015abe0, 0xc00043b020)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /go/pkg/mod/git.mills.io/prologic/msgbus@v0.1.13/queue.go:126 +0x110
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | git.mills.io/prologic/msgbus.(*MessageBus).Subscribe(0xc0004fa000, {0xc0003088f0, 0xf}, {0xc0003aae05, 0x20}, {0xc0005330a8, 0x1, 0x40d987?})
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /go/pkg/mod/git.mills.io/prologic/msgbus@v0.1.13/msgbus.go:486 +0x5ad
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | git.mills.io/prologic/msgbus.(*Client).Start(0xc0000ac300)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /go/pkg/mod/git.mills.io/prologic/msgbus@v0.1.13/msgbus.go:742 +0xb5
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | git.mills.io/prologic/msgbus.(*MessageBus).ServeHTTP(0xc0004fa000, {0x2b1ea08, 0xc0003b4000}, 0xc0002a8200)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /go/pkg/mod/git.mills.io/prologic/msgbus@v0.1.13/msgbus.go:617 +0x758
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | go.mills.io/saltyim/internal.(*Server).initRoutes.func1({0x2b1ea08?, 0xc0003b4000?}, 0xd8732a?, {0x1d?, 0x0?, 0x0?})
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /src/internal/handlers.go:53 +0x37
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | go.mills.io/saltyim/internal.NewServer.func1.1({0x2b1ea08, 0xc0003b4000}, 0xc0003aa200?, {0xc0000ca160, 0x1, 0x1})
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /src/internal/server.go:388 +0x26d
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | github.com/julienschmidt/httprouter.(*Router).ServeHTTP(0xc00024e900, {0x2b1ea08, 0xc0003b4000}, 0xc0002a8200)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /go/pkg/mod/github.com/julienschmidt/httprouter@v1.3.0/router.go:387 +0x82b
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | go.mills.io/saltyim/internal.(*Router).ServeHTTP(0xc44960?, {0x2b1ea08?, 0xc0003b4000?}, 0x4?)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /src/internal/router.go:144 +0x25
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | github.com/justinas/nosurf.(*CSRFHandler).handleSuccess(...)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /go/pkg/mod/github.com/justinas/nosurf@v1.1.1/handler.go:187
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | github.com/justinas/nosurf.(*CSRFHandler).ServeHTTP(0xc0002245a0, {0x2b1ea08, 0xc0003b4000}, 0x4?)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /go/pkg/mod/github.com/justinas/nosurf@v1.1.1/handler.go:144 +0x5c4
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | github.com/NYTimes/gziphandler.GzipHandlerWithOpts.func1.1({0x2b1f0f8, 0xc0000ca060}, 0x4c4801?)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /go/pkg/mod/github.com/!n!y!times/gziphandler@v1.1.1/gzip.go:338 +0x262
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | net/http.HandlerFunc.ServeHTTP(0x2?, {0x2b1f0f8?, 0xc0000ca060?}, 0xc0003d7380?)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /usr/local/go/src/net/http/server.go:2084 +0x2f
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | github.com/unrolled/logger.(*Logger).Handler.func1({0x2b1f7e8?, 0xc0001980e0}, 0xc0002a8100)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /go/pkg/mod/github.com/unrolled/logger@v0.0.0-20201216141554-31a3694fe979/logger.go:80 +0xf4
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | net/http.HandlerFunc.ServeHTTP(0xc0003aa226?, {0x2b1f7e8?, 0xc0001980e0?}, 0xc000188400?)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /usr/local/go/src/net/http/server.go:2084 +0x2f
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | net/http.serverHandler.ServeHTTP({0xc00007c210?}, {0x2b1f7e8, 0xc0001980e0}, 0xc0002a8100)
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /usr/local/go/src/net/http/server.go:2916 +0x43b
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | net/http.(*conn).serve(0xc000790000, {0x2b20228, 0xc0002ab830})
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /usr/local/go/src/net/http/server.go:1966 +0x5d7
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | created by net/http.(*Server).Serve
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | /usr/local/go/src/net/http/server.go:3071 +0x4db
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | time="2022-04-02T21:49:52Z" level=debug msg="pong latency of 10.0.2.82:49826: 1.671603ms"
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | time="2022-04-02T21:50:20Z" level=info msg="synced store"
saltyim_saltyd.1.w8s8tw903qnj@dm3.mills.io | [saltyd] 2022/04/02 21:50:32 (172.69.42.134) "GET /avatar/d3d52221e8da5a8ae012f4e2db0631c181f4156f0edcde5cffa25b347c7ceda8 HTTP/1.1" 304 0 108.078µs
```
Co-authored-by: James Mills <prologic@shortcircuit.net.au>
Reviewed-on: https://git.mills.io/prologic/msgbus/pulls/29
2022-04-02 21:59:10 +00:00
|
|
|
if msg, ok := item.(Message); ok && msg.ID >= o.Index {
|
2022-04-04 01:15:49 +00:00
|
|
|
log.Debugf("found msg %s#%d", msg.Topic.Name, msg.ID)
|
|
|
|
select {
|
|
|
|
case ch <- msg:
|
|
|
|
n++
|
|
|
|
default:
|
2022-04-05 06:09:34 +00:00
|
|
|
if mb.metrics != nil {
|
|
|
|
mb.metrics.Counter("bus", "dropped").Inc()
|
|
|
|
}
|
2022-04-04 01:15:49 +00:00
|
|
|
return ErrBufferFull
|
|
|
|
}
|
2022-04-02 14:05:15 +00:00
|
|
|
}
|
|
|
|
return nil
|
|
|
|
})
|
2022-04-04 01:15:49 +00:00
|
|
|
if err != nil {
|
|
|
|
log.WithError(err).Error("error publishing messages to new subscriber")
|
|
|
|
}
|
2022-04-02 14:05:15 +00:00
|
|
|
log.Debugf("published %d messages", n)
|
2022-04-04 02:36:56 +00:00
|
|
|
return ch
|
2022-04-02 14:05:15 +00:00
|
|
|
}
|
|
|
|
|
2022-04-04 02:36:56 +00:00
|
|
|
// Otherwise, s.Index was eitehr 0 (start from head)
|
|
|
|
// OR > topic.Sequence in which case (start from head)
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
return ch
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// Unsubscribe ...
|
|
|
|
func (mb *MessageBus) Unsubscribe(id, topic string) {
|
2018-05-10 09:46:25 +00:00
|
|
|
mb.Lock()
|
|
|
|
defer mb.Unlock()
|
|
|
|
|
2017-08-14 07:34:12 +00:00
|
|
|
t, ok := mb.topics[topic]
|
|
|
|
if !ok {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
subs, ok := mb.subscribers[t]
|
2017-06-03 15:16:17 +00:00
|
|
|
if !ok {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
if subs.HasSubscriber(id) {
|
2018-05-08 07:18:34 +00:00
|
|
|
// Already verified the listener exists
|
2022-04-02 14:05:15 +00:00
|
|
|
subs.RemoveSubscriber(id)
|
2018-05-08 07:18:34 +00:00
|
|
|
|
|
|
|
if mb.metrics != nil {
|
|
|
|
mb.metrics.Gauge("bus", "subscribers").Dec()
|
|
|
|
}
|
2017-06-03 15:16:17 +00:00
|
|
|
}
|
|
|
|
}
|
2017-08-09 09:54:11 +00:00
|
|
|
|
|
|
|
func (mb *MessageBus) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
2018-05-02 07:41:14 +00:00
|
|
|
defer func() {
|
2018-05-02 08:24:30 +00:00
|
|
|
if mb.metrics != nil {
|
2018-05-05 08:29:08 +00:00
|
|
|
mb.metrics.Counter("server", "requests").Inc()
|
2018-05-02 08:24:30 +00:00
|
|
|
}
|
2018-05-02 07:41:14 +00:00
|
|
|
}()
|
|
|
|
|
2022-03-25 01:27:23 +00:00
|
|
|
w.Header().Set("Access-Control-Allow-Origin", "*")
|
2022-03-29 22:03:38 +00:00
|
|
|
w.Header().Set("Accept-Encoding", "br, gzip, deflate")
|
2022-03-25 01:27:23 +00:00
|
|
|
|
2017-08-09 09:54:11 +00:00
|
|
|
if r.Method == "GET" && (r.URL.Path == "/" || r.URL.Path == "") {
|
2018-04-06 08:27:25 +00:00
|
|
|
// XXX: guard with a mutex?
|
2018-03-25 23:27:39 +00:00
|
|
|
out, err := json.Marshal(mb.topics)
|
|
|
|
if err != nil {
|
2018-04-06 08:27:25 +00:00
|
|
|
msg := fmt.Sprintf("error serializing topics: %s", err)
|
|
|
|
http.Error(w, msg, http.StatusInternalServerError)
|
2018-03-25 23:27:39 +00:00
|
|
|
return
|
2017-08-09 09:54:11 +00:00
|
|
|
}
|
2018-03-25 23:27:39 +00:00
|
|
|
|
2018-05-14 10:04:45 +00:00
|
|
|
w.WriteHeader(http.StatusOK)
|
2018-03-25 23:27:39 +00:00
|
|
|
w.Header().Set("Content-Type", "application/json")
|
|
|
|
w.Write(out)
|
2017-08-09 09:54:11 +00:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2022-04-02 14:05:15 +00:00
|
|
|
topic := strings.Trim(r.URL.Path, "/")
|
2017-08-09 09:54:11 +00:00
|
|
|
|
2018-04-02 21:38:33 +00:00
|
|
|
t := mb.NewTopic(topic)
|
2022-04-03 15:59:38 +00:00
|
|
|
log.Debugf("request for topic %s", t.Name)
|
2017-08-14 07:34:12 +00:00
|
|
|
|
2017-08-09 09:54:11 +00:00
|
|
|
switch r.Method {
|
|
|
|
case "POST", "PUT":
|
2022-03-29 22:03:38 +00:00
|
|
|
defer r.Body.Close()
|
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
if r.ContentLength > int64(mb.options.MaxPayloadSize) {
|
2018-05-10 06:25:13 +00:00
|
|
|
msg := "payload exceeds max-payload-size"
|
|
|
|
http.Error(w, msg, http.StatusRequestEntityTooLarge)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2022-03-29 22:03:38 +00:00
|
|
|
var rd io.Reader = r.Body
|
|
|
|
ce := r.Header.Get("Content-Encoding")
|
|
|
|
switch ce {
|
|
|
|
case "":
|
|
|
|
case "br":
|
|
|
|
rd = brotli.NewReader(rd)
|
|
|
|
case "gzip":
|
|
|
|
gz, err := gzip.NewReader(rd)
|
|
|
|
if err != nil {
|
|
|
|
msg := fmt.Sprintf("error reading payload: %s", err)
|
|
|
|
http.Error(w, msg, http.StatusBadRequest)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
defer gz.Close()
|
|
|
|
rd = gz
|
|
|
|
case "deflate":
|
|
|
|
fl := flate.NewReader(rd)
|
|
|
|
defer fl.Close()
|
|
|
|
rd = fl
|
|
|
|
default:
|
|
|
|
msg := fmt.Sprintf("error reading payload: not acceptable: %v", ce)
|
|
|
|
http.Error(w, msg, http.StatusNotAcceptable)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
body, err := ioutil.ReadAll(rd)
|
2017-08-09 09:54:11 +00:00
|
|
|
if err != nil {
|
2018-04-06 08:27:25 +00:00
|
|
|
msg := fmt.Sprintf("error reading payload: %s", err)
|
|
|
|
http.Error(w, msg, http.StatusBadRequest)
|
2017-08-09 09:54:11 +00:00
|
|
|
return
|
|
|
|
}
|
2022-04-03 15:59:38 +00:00
|
|
|
if len(body) > mb.options.MaxPayloadSize {
|
2018-05-10 06:25:13 +00:00
|
|
|
msg := "payload exceeds max-payload-size"
|
|
|
|
http.Error(w, msg, http.StatusRequestEntityTooLarge)
|
|
|
|
return
|
|
|
|
}
|
2018-05-14 10:04:45 +00:00
|
|
|
mb.Put(mb.NewMessage(t, body))
|
2018-04-06 08:27:25 +00:00
|
|
|
|
2018-05-14 10:04:45 +00:00
|
|
|
w.WriteHeader(http.StatusAccepted)
|
2017-08-09 09:54:11 +00:00
|
|
|
case "GET":
|
|
|
|
if r.Header.Get("Upgrade") == "websocket" {
|
2022-03-27 02:42:56 +00:00
|
|
|
conn, err := upgrader.Upgrade(w, r, nil)
|
2018-05-02 06:19:04 +00:00
|
|
|
if err != nil {
|
|
|
|
log.Errorf("error creating websocket client: %s", err)
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2022-04-03 15:59:38 +00:00
|
|
|
i := SafeParseInt64(r.URL.Query().Get("index"), -1)
|
2022-04-02 14:05:15 +00:00
|
|
|
|
|
|
|
log.Debugf("new subscriber for %s from %s", t.Name, r.RemoteAddr)
|
|
|
|
|
|
|
|
NewClient(conn, t, i, mb).Start()
|
2017-08-09 09:54:11 +00:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2017-08-14 07:34:12 +00:00
|
|
|
message, ok := mb.Get(t)
|
2017-08-09 09:54:11 +00:00
|
|
|
|
|
|
|
if !ok {
|
2022-03-31 10:55:03 +00:00
|
|
|
http.Error(w, "No Messages", http.StatusNoContent)
|
2017-08-09 09:54:11 +00:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
out, err := json.Marshal(message)
|
|
|
|
if err != nil {
|
2018-04-06 08:27:25 +00:00
|
|
|
msg := fmt.Sprintf("error serializing message: %s", err)
|
|
|
|
http.Error(w, msg, http.StatusInternalServerError)
|
2017-08-09 09:54:11 +00:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2018-05-14 10:04:45 +00:00
|
|
|
w.WriteHeader(http.StatusOK)
|
2018-03-25 23:27:39 +00:00
|
|
|
w.Header().Set("Content-Type", "application/json")
|
2017-08-09 09:54:11 +00:00
|
|
|
w.Write(out)
|
|
|
|
case "DELETE":
|
|
|
|
http.Error(w, "Not Implemented", http.StatusNotImplemented)
|
2018-05-02 07:41:14 +00:00
|
|
|
// TODO: Implement deleting topics
|
2017-08-09 09:54:11 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-08-14 07:34:12 +00:00
|
|
|
// Client ...
|
2017-08-09 09:54:11 +00:00
|
|
|
type Client struct {
|
2018-05-02 06:19:04 +00:00
|
|
|
conn *websocket.Conn
|
2017-08-14 07:34:12 +00:00
|
|
|
topic *Topic
|
2022-04-03 15:59:38 +00:00
|
|
|
index int64
|
2017-08-09 09:54:11 +00:00
|
|
|
bus *MessageBus
|
2018-05-02 06:19:04 +00:00
|
|
|
|
|
|
|
id string
|
|
|
|
ch chan Message
|
2017-08-09 09:54:11 +00:00
|
|
|
}
|
|
|
|
|
2017-08-14 07:34:12 +00:00
|
|
|
// NewClient ...
|
2022-04-03 15:59:38 +00:00
|
|
|
func NewClient(conn *websocket.Conn, topic *Topic, index int64, bus *MessageBus) *Client {
|
2022-04-04 01:56:56 +00:00
|
|
|
return &Client{
|
|
|
|
conn: conn,
|
|
|
|
topic: topic,
|
|
|
|
index: index,
|
|
|
|
bus: bus,
|
|
|
|
|
|
|
|
id: MustGenerateULID(),
|
|
|
|
}
|
2017-08-09 09:54:11 +00:00
|
|
|
}
|
|
|
|
|
2022-03-27 02:42:56 +00:00
|
|
|
func (c *Client) readPump() {
|
|
|
|
defer func() {
|
|
|
|
c.conn.Close()
|
|
|
|
}()
|
|
|
|
|
|
|
|
c.conn.SetReadDeadline(time.Now().Add(pongWait))
|
|
|
|
|
|
|
|
c.conn.SetPongHandler(func(message string) error {
|
|
|
|
t, err := strconv.ParseInt(message, 10, 64)
|
|
|
|
d := time.Duration(time.Now().UnixNano() - t)
|
|
|
|
if err != nil {
|
|
|
|
log.Warnf("garbage pong reply from %s: %s", c.id, err)
|
|
|
|
} else {
|
|
|
|
log.Debugf("pong latency of %s: %s", c.id, d)
|
|
|
|
}
|
|
|
|
c.conn.SetReadDeadline(time.Now().Add(pongWait))
|
|
|
|
|
|
|
|
if c.bus.metrics != nil {
|
|
|
|
v := c.bus.metrics.Summary("client", "latency_seconds")
|
|
|
|
v.Observe(d.Seconds())
|
|
|
|
}
|
2018-05-05 08:29:08 +00:00
|
|
|
|
2022-03-27 02:42:56 +00:00
|
|
|
return nil
|
|
|
|
})
|
2017-08-09 09:54:11 +00:00
|
|
|
|
2018-05-02 06:19:04 +00:00
|
|
|
for {
|
2022-03-27 02:42:56 +00:00
|
|
|
_, message, err := c.conn.ReadMessage()
|
|
|
|
if err != nil {
|
|
|
|
c.bus.Unsubscribe(c.id, c.topic.Name)
|
2022-03-20 15:10:02 +00:00
|
|
|
return
|
2018-05-03 07:57:52 +00:00
|
|
|
}
|
2022-03-27 02:42:56 +00:00
|
|
|
log.Debugf("recieved message from %s: %s", c.id, message)
|
2018-05-03 07:57:52 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2022-03-27 02:42:56 +00:00
|
|
|
func (c *Client) writePump() {
|
|
|
|
ticker := time.NewTicker(pingPeriod)
|
|
|
|
defer func() {
|
|
|
|
ticker.Stop()
|
|
|
|
c.conn.Close()
|
|
|
|
}()
|
|
|
|
|
|
|
|
var err error
|
|
|
|
|
2018-05-03 07:57:52 +00:00
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case msg, ok := <-c.ch:
|
2022-03-27 02:42:56 +00:00
|
|
|
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
|
2018-05-03 07:57:52 +00:00
|
|
|
if !ok {
|
2022-03-27 02:42:56 +00:00
|
|
|
// The bus closed the channel.
|
|
|
|
message := websocket.FormatCloseMessage(websocket.CloseNormalClosure, "bus closed")
|
|
|
|
c.conn.WriteControl(websocket.CloseMessage, message, time.Now().Add(writeWait))
|
2018-05-03 07:57:52 +00:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2022-03-27 02:42:56 +00:00
|
|
|
err = c.conn.WriteJSON(msg)
|
2018-05-03 07:57:52 +00:00
|
|
|
if err != nil {
|
|
|
|
// TODO: Retry? Put the message back in the queue?
|
2018-05-06 16:34:52 +00:00
|
|
|
log.Errorf("Error sending msg to %s: %s", c.id, err)
|
2018-05-03 07:57:52 +00:00
|
|
|
if c.bus.metrics != nil {
|
2018-05-05 08:29:08 +00:00
|
|
|
c.bus.metrics.Counter("client", "errors").Inc()
|
2018-05-03 07:57:52 +00:00
|
|
|
}
|
|
|
|
} else {
|
|
|
|
if c.bus.metrics != nil {
|
|
|
|
c.bus.metrics.Counter("bus", "delivered").Inc()
|
|
|
|
}
|
|
|
|
}
|
2022-03-27 02:42:56 +00:00
|
|
|
case <-ticker.C:
|
|
|
|
c.conn.SetWriteDeadline(time.Now().Add(writeWait))
|
|
|
|
t := time.Now()
|
|
|
|
message := []byte(fmt.Sprintf("%d", t.UnixNano()))
|
|
|
|
if err := c.conn.WriteMessage(websocket.PingMessage, message); err != nil {
|
|
|
|
log.Errorf("error sending ping to %s: %s", c.id, err)
|
|
|
|
return
|
|
|
|
}
|
2017-08-09 09:54:11 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2018-05-03 07:57:52 +00:00
|
|
|
|
|
|
|
// Start ...
|
|
|
|
func (c *Client) Start() {
|
2022-04-02 14:05:15 +00:00
|
|
|
c.ch = c.bus.Subscribe(c.id, c.topic.Name, WithIndex(c.index))
|
2018-05-03 07:57:52 +00:00
|
|
|
|
2022-03-27 02:42:56 +00:00
|
|
|
c.conn.SetCloseHandler(func(code int, text string) error {
|
|
|
|
c.bus.Unsubscribe(c.id, c.topic.Name)
|
|
|
|
message := websocket.FormatCloseMessage(code, text)
|
|
|
|
c.conn.WriteControl(websocket.CloseMessage, message, time.Now().Add(writeWait))
|
|
|
|
return nil
|
|
|
|
})
|
2018-05-03 07:57:52 +00:00
|
|
|
|
2022-03-27 02:42:56 +00:00
|
|
|
go c.writePump()
|
|
|
|
go c.readPump()
|
2018-05-03 07:57:52 +00:00
|
|
|
}
|