add dispatcher
simplify transport implementation support wildcard subscriptions
This commit is contained in:
Marcelo Pires
269
internal/dispatcher/dispatcher.go
Normal file
269
internal/dispatcher/dispatcher.go
Normal file
@ -0,0 +1,269 @@
|
||||
package dispatcher
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/thesyncim/faye/internal/store"
|
||||
"github.com/thesyncim/faye/message"
|
||||
"github.com/thesyncim/faye/subscription"
|
||||
"github.com/thesyncim/faye/transport"
|
||||
"log"
|
||||
"strconv"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
type dispatcher interface {
|
||||
SetTransport(t transport.Transport)
|
||||
//Subscribe informs the server that messages published to that channel are delivered to itself.
|
||||
Subscribe(channel string) (*subscription.Subscription, error)
|
||||
//Unsubscribe informs the server that the client will no longer listen to incoming event messages on
|
||||
//the specified channel/subscription
|
||||
Unsubscribe(sub *subscription.Subscription) error
|
||||
//Publish publishes events on a channel by sending event messages, the server MAY respond to a publish event
|
||||
//if this feature is supported by the server use the OnPublishResponse to get the publish status.
|
||||
Publish(subscription string, message message.Data) (id string, err error)
|
||||
//OnPublishResponse sets the handler to be triggered if the server replies to the publish request
|
||||
//according to the spec the server MAY reply to the publish request, so its not guaranteed that this handler will
|
||||
//ever be triggered
|
||||
//can be used to identify the status of the published request and for example retry failed published requests
|
||||
OnPublishResponse(subscription string, onMsg func(message *message.Message))
|
||||
}
|
||||
|
||||
var _ dispatcher = (*Dispatcher)(nil)
|
||||
|
||||
type Dispatcher struct {
|
||||
endpoint string
|
||||
//transports map[string]transport.Transport
|
||||
transport transport.Transport
|
||||
transportOpts transport.Options
|
||||
|
||||
msgID *uint64
|
||||
|
||||
extensions message.Extensions
|
||||
|
||||
//map requestID
|
||||
pendingSubs map[string]chan error //todo wrap in structure
|
||||
pendingSubsMu sync.Mutex
|
||||
store *store.SubscriptionsStore
|
||||
|
||||
onPublishResponseMu sync.Mutex //todo sync.Map
|
||||
onPublishResponse map[string]func(message *message.Message)
|
||||
|
||||
clientID string
|
||||
}
|
||||
|
||||
func NewDispatcher(endpoint string, tOpts transport.Options, ext message.Extensions) *Dispatcher {
|
||||
var msgID uint64
|
||||
return &Dispatcher{
|
||||
endpoint: endpoint,
|
||||
msgID: &msgID,
|
||||
store: store.NewStore(100),
|
||||
transportOpts: tOpts,
|
||||
extensions: ext,
|
||||
onPublishResponse: map[string]func(message *message.Message){},
|
||||
pendingSubs: map[string]chan error{},
|
||||
}
|
||||
}
|
||||
|
||||
//todo allow multiple transports
|
||||
func (d *Dispatcher) Connect() error {
|
||||
var err error
|
||||
if err = d.transport.Init(d.endpoint, &d.transportOpts); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err = d.metaHandshake(); err != nil {
|
||||
return err
|
||||
}
|
||||
return d.metaConnect()
|
||||
}
|
||||
|
||||
func (d *Dispatcher) metaHandshake() error {
|
||||
m := &message.Message{
|
||||
Channel: message.MetaHandshake,
|
||||
Version: "1.0", //todo const
|
||||
SupportedConnectionTypes: []string{d.transport.Name()}, //todo list all tranports
|
||||
}
|
||||
d.extensions.ApplyOutExtensions(m)
|
||||
handshakeResp, err := d.transport.Handshake(m)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
d.extensions.ApplyInExtensions(handshakeResp)
|
||||
if handshakeResp.GetError() != nil {
|
||||
return err
|
||||
}
|
||||
d.clientID = handshakeResp.ClientId
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *Dispatcher) metaConnect() error {
|
||||
m := &message.Message{
|
||||
Channel: message.MetaConnect,
|
||||
ClientId: d.clientID,
|
||||
ConnectionType: d.transport.Name(),
|
||||
Id: d.nextMsgID(),
|
||||
}
|
||||
return d.transport.Connect(m)
|
||||
}
|
||||
|
||||
func (d *Dispatcher) Disconnect() error {
|
||||
m := &message.Message{
|
||||
Channel: message.MetaDisconnect,
|
||||
ClientId: d.clientID,
|
||||
Id: d.nextMsgID(),
|
||||
}
|
||||
return d.transport.Disconnect(m)
|
||||
}
|
||||
|
||||
func (d *Dispatcher) dispatchMessage(msg *message.Message) {
|
||||
d.extensions.ApplyInExtensions(msg)
|
||||
|
||||
if message.IsMetaMessage(msg) {
|
||||
//handle it
|
||||
switch msg.Channel {
|
||||
case message.MetaSubscribe:
|
||||
//handle MetaSubscribe resp
|
||||
d.pendingSubsMu.Lock()
|
||||
confirmCh, ok := d.pendingSubs[msg.Id]
|
||||
d.pendingSubsMu.Unlock()
|
||||
if !ok {
|
||||
panic("BUG: subscription not registered `" + msg.Subscription + "`")
|
||||
}
|
||||
|
||||
if !msg.Successful {
|
||||
if msg.GetError() == nil {
|
||||
//inject the error if the server returns unsuccessful without error
|
||||
msg.Error = fmt.Sprintf("susbscription `%s` failed", msg.Subscription)
|
||||
}
|
||||
confirmCh <- msg.GetError()
|
||||
//v2
|
||||
} else {
|
||||
confirmCh <- nil
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
//is Event Message
|
||||
//there are 2 types of Event Message
|
||||
// 1. Publish
|
||||
// 2. Delivery
|
||||
if message.IsEventDelivery(msg) {
|
||||
subscriptions := d.store.Match(msg.Channel)
|
||||
//send to all listeners
|
||||
log.Println(subscriptions, msg)
|
||||
for i := range subscriptions {
|
||||
if subscriptions[i].MsgChannel() != nil {
|
||||
select {
|
||||
case subscriptions[i].MsgChannel() <- msg:
|
||||
default:
|
||||
log.Println("subscription has no listeners") //todo remove
|
||||
}
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if message.IsEventPublish(msg) {
|
||||
d.onPublishResponseMu.Lock()
|
||||
onPublish, ok := d.onPublishResponse[msg.Channel]
|
||||
d.onPublishResponseMu.Unlock()
|
||||
if ok {
|
||||
onPublish(msg)
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (d *Dispatcher) SetTransport(t transport.Transport) {
|
||||
t.SetOnMessageReceivedHandler(d.dispatchMessage)
|
||||
d.transport = t
|
||||
}
|
||||
|
||||
func (d *Dispatcher) nextMsgID() string {
|
||||
return strconv.Itoa(int(atomic.AddUint64(d.msgID, 1)))
|
||||
}
|
||||
|
||||
//sendMessage send applies the out extensions and sends a message throught the transport
|
||||
func (d *Dispatcher) sendMessage(m *message.Message) error {
|
||||
d.extensions.ApplyOutExtensions(m)
|
||||
return d.transport.SendMessage(m)
|
||||
}
|
||||
|
||||
func (d *Dispatcher) Subscribe(channel string) (*subscription.Subscription, error) {
|
||||
id := d.nextMsgID()
|
||||
m := &message.Message{
|
||||
Channel: message.MetaSubscribe,
|
||||
ClientId: d.clientID,
|
||||
Subscription: channel,
|
||||
Id: id,
|
||||
}
|
||||
|
||||
if err := d.transport.SendMessage(m); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
inMsgCh := make(chan *message.Message, 0)
|
||||
subscriptionConfirmation := make(chan error)
|
||||
|
||||
d.pendingSubsMu.Lock()
|
||||
d.pendingSubs[id] = subscriptionConfirmation
|
||||
d.pendingSubsMu.Unlock()
|
||||
|
||||
sub, err := subscription.NewSubscription(channel, d.Unsubscribe, inMsgCh)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
//todo timeout here
|
||||
err = <-subscriptionConfirmation
|
||||
if err != nil {
|
||||
//log.Println(err)
|
||||
return nil, err
|
||||
}
|
||||
d.store.Add(sub)
|
||||
return sub, nil
|
||||
|
||||
}
|
||||
|
||||
func (d *Dispatcher) Unsubscribe(sub *subscription.Subscription) error {
|
||||
//https://docs.cometd.org/current/reference/#_bayeux_meta_unsubscribe
|
||||
d.store.Remove(sub)
|
||||
//if this is last subscription we will send meta unsubscribe to the server
|
||||
if d.store.Count(sub.Name()) == 0 {
|
||||
d.onPublishResponseMu.Lock()
|
||||
delete(d.onPublishResponse, sub.Name())
|
||||
d.onPublishResponseMu.Unlock()
|
||||
|
||||
m := &message.Message{
|
||||
Channel: message.MetaUnsubscribe,
|
||||
Subscription: sub.Name(),
|
||||
ClientId: d.clientID,
|
||||
Id: d.nextMsgID(),
|
||||
}
|
||||
return d.transport.SendMessage(m)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *Dispatcher) Publish(subscription string, data message.Data) (id string, err error) {
|
||||
id = d.nextMsgID()
|
||||
|
||||
m := &message.Message{
|
||||
Channel: subscription,
|
||||
Data: data,
|
||||
ClientId: d.clientID,
|
||||
Id: id,
|
||||
}
|
||||
if err = d.sendMessage(m); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
|
||||
func (d *Dispatcher) OnPublishResponse(subscription string, onMsg func(message *message.Message)) {
|
||||
d.onPublishResponseMu.Lock()
|
||||
d.onPublishResponse[subscription] = onMsg
|
||||
d.onPublishResponseMu.Unlock()
|
||||
}
|
||||
40
internal/store/name.go
Normal file
40
internal/store/name.go
Normal file
@ -0,0 +1,40 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"strings"
|
||||
)
|
||||
|
||||
type Name struct {
|
||||
n string
|
||||
patterns []string
|
||||
}
|
||||
|
||||
func NewName(name string) *Name {
|
||||
var n Name
|
||||
n.n = name
|
||||
//expand once
|
||||
n.patterns = n.expand()
|
||||
return &n
|
||||
}
|
||||
|
||||
func (n *Name) Match(channel string) bool {
|
||||
for i := range n.patterns {
|
||||
if n.patterns[i] == channel {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (n *Name) expand() []string {
|
||||
segments := strings.Split(n.n, "/")
|
||||
num_segments := len(segments)
|
||||
patterns := make([]string, num_segments+1)
|
||||
patterns[0] = "/**"
|
||||
for i := 1; i < len(segments); i = i + 2 {
|
||||
patterns[i] = strings.Join(segments[:i+1], "/") + "/**"
|
||||
}
|
||||
patterns[len(patterns)-2] = strings.Join(segments[:num_segments-1], "/") + "/*"
|
||||
patterns[len(patterns)-1] = n.n
|
||||
return patterns
|
||||
}
|
||||
90
internal/store/subscription.go
Normal file
90
internal/store/subscription.go
Normal file
@ -0,0 +1,90 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"github.com/thesyncim/faye/subscription"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type SubscriptionsStore struct {
|
||||
mutex sync.Mutex
|
||||
subs map[string][]*subscription.Subscription
|
||||
|
||||
//cache for expanded channel names
|
||||
cache map[string]*Name
|
||||
}
|
||||
|
||||
func NewStore(size int) *SubscriptionsStore {
|
||||
return &SubscriptionsStore{
|
||||
subs: make(map[string][]*subscription.Subscription, size),
|
||||
cache: map[string]*Name{},
|
||||
}
|
||||
}
|
||||
|
||||
func (s *SubscriptionsStore) Add(sub *subscription.Subscription) {
|
||||
s.mutex.Lock()
|
||||
s.subs[sub.Name()] = append(s.subs[sub.Name()], sub)
|
||||
s.mutex.Unlock()
|
||||
}
|
||||
|
||||
//Match returns the subscriptions that match with the specified channel name
|
||||
//Wildcard subscriptions are matched
|
||||
func (s *SubscriptionsStore) Match(channel string) []*subscription.Subscription {
|
||||
var (
|
||||
matches []*subscription.Subscription
|
||||
name *Name
|
||||
ok bool
|
||||
)
|
||||
s.mutex.Lock()
|
||||
if name, ok = s.cache[channel]; !ok {
|
||||
name = NewName(channel)
|
||||
s.cache[channel] = name
|
||||
}
|
||||
for _, subs := range s.subs {
|
||||
for i := range subs {
|
||||
if name.Match(subs[i].Name()) {
|
||||
matches = append(matches, subs[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
s.mutex.Unlock()
|
||||
return matches
|
||||
}
|
||||
|
||||
func (s *SubscriptionsStore) Remove(sub *subscription.Subscription) {
|
||||
close(sub.MsgChannel())
|
||||
s.mutex.Lock()
|
||||
for channel, subs := range s.subs {
|
||||
for i := range subs {
|
||||
if subs[i] == sub {
|
||||
//delete the subscription
|
||||
subs = subs[:i+copy(subs[i:], subs[i+1:])]
|
||||
if len(subs) == 0 {
|
||||
delete(s.subs, channel)
|
||||
}
|
||||
goto end
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
end:
|
||||
s.mutex.Unlock()
|
||||
}
|
||||
|
||||
//RemoveAll removel all subscriptions and close all channels
|
||||
func (s *SubscriptionsStore) RemoveAll() {
|
||||
s.mutex.Lock()
|
||||
for i := range s.subs {
|
||||
//close all listeners
|
||||
for j := range s.subs[i] {
|
||||
s.subs[i][j].Unsubscribe()
|
||||
close(s.subs[i][j].MsgChannel())
|
||||
}
|
||||
delete(s.subs, i)
|
||||
}
|
||||
s.mutex.Unlock()
|
||||
}
|
||||
|
||||
//Count return the number of subscriptions associated with the specified channel
|
||||
func (s *SubscriptionsStore) Count(channel string) int {
|
||||
return len(s.Match(channel))
|
||||
}
|
||||
140
internal/store/subscription_test.go
Normal file
140
internal/store/subscription_test.go
Normal file
@ -0,0 +1,140 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"github.com/thesyncim/faye/subscription"
|
||||
"reflect"
|
||||
"testing"
|
||||
)
|
||||
|
||||
var (
|
||||
wildcardSubscription, _ = subscription.NewSubscription("a", "/wildcard/*", nil, nil, nil)
|
||||
simpleSubscription, _ = subscription.NewSubscription("b", "/foo/bar", nil, nil, nil)
|
||||
)
|
||||
|
||||
func TestStore_Add(t *testing.T) {
|
||||
type args struct {
|
||||
name string
|
||||
subs []*subscription.Subscription
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
s *SubscriptionsStore
|
||||
args args
|
||||
expected *SubscriptionsStore
|
||||
}{
|
||||
|
||||
{
|
||||
name: "add one",
|
||||
s: NewStore(10),
|
||||
args: args{
|
||||
name: "/wildcard/*",
|
||||
subs: []*subscription.Subscription{wildcardSubscription},
|
||||
},
|
||||
expected: &SubscriptionsStore{
|
||||
subs: map[string][]*subscription.Subscription{
|
||||
"/wildcard/*": {wildcardSubscription},
|
||||
},
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "add three",
|
||||
s: NewStore(10),
|
||||
args: args{
|
||||
name: "/wildcard/*",
|
||||
subs: []*subscription.Subscription{wildcardSubscription, wildcardSubscription, wildcardSubscription},
|
||||
},
|
||||
expected: &SubscriptionsStore{
|
||||
subs: map[string][]*subscription.Subscription{
|
||||
"/wildcard/*": {wildcardSubscription, wildcardSubscription, wildcardSubscription},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
|
||||
for i := range tt.args.subs {
|
||||
tt.s.Add(tt.args.subs[i])
|
||||
}
|
||||
|
||||
if !reflect.DeepEqual(tt.expected, tt.s) {
|
||||
t.Fatalf("expecting :%v got: %v", tt.expected, tt.s)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStore_Match(t *testing.T) {
|
||||
type args struct {
|
||||
name string
|
||||
}
|
||||
simpleStore := NewStore(0)
|
||||
simpleStore.Add(simpleSubscription)
|
||||
wildcardStore := NewStore(0)
|
||||
wildcardStore.Add(wildcardSubscription)
|
||||
tests := []struct {
|
||||
name string
|
||||
s *SubscriptionsStore
|
||||
args args
|
||||
want []*subscription.Subscription
|
||||
}{
|
||||
{
|
||||
name: "match simple",
|
||||
s: simpleStore,
|
||||
want: []*subscription.Subscription{simpleSubscription},
|
||||
args: args{
|
||||
name: "/foo/bar",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "match wildcard 1",
|
||||
s: wildcardStore,
|
||||
want: []*subscription.Subscription{wildcardSubscription},
|
||||
args: args{
|
||||
name: "/wildcard/a",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "match wildcard 2",
|
||||
s: wildcardStore,
|
||||
want: []*subscription.Subscription{wildcardSubscription},
|
||||
args: args{
|
||||
name: "/wildcard/ccc",
|
||||
},
|
||||
},
|
||||
{
|
||||
name: "match non existent",
|
||||
s: wildcardStore,
|
||||
want: nil,
|
||||
args: args{
|
||||
name: "/wildcardsadasd",
|
||||
},
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
if got := tt.s.Match(tt.args.name); !reflect.DeepEqual(got, tt.want) {
|
||||
t.Errorf("SubscriptionsStore.Match() = %#v, want %#v", got, tt.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestStore_Remove(t *testing.T) {
|
||||
type args struct {
|
||||
sub *subscription.Subscription
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
s *SubscriptionsStore
|
||||
args args
|
||||
}{
|
||||
// TODO: Add test cases.
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
tt.s.Remove(tt.args.sub)
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user