added query tests

broker/cluster/swarm.go

@@ -39,7 +39,7 @@ type Swarm struct {
 	state   *subscriptionState    // The state to synchronise.
 	router  *mesh.Router          // The mesh router.
 	gossip  mesh.Gossip           // The gossip protocol.
-	members *sync.Map             // The map of members in the peer set.
+	members sync.Map              // The map of members in the peer set.
 
 	OnSubscribe   func(message.Ssid, message.Subscriber) bool // Delegate to invoke when the subscription event is received.
 	OnUnsubscribe func(message.Ssid, message.Subscriber) bool // Delegate to invoke when the subscription event is received.
@@ -57,7 +57,6 @@ func NewSwarm(cfg *config.ClusterConfig, closing chan bool) *Swarm {
 		closing: closing,
 		config:  cfg,
 		state:   newSubscriptionState(),
-		members: new(sync.Map),
 	}
 
 	// Get the cluster binding address
@@ -215,6 +214,10 @@ func (s *Swarm) merge(buf []byte) (mesh.GossipData, error) {
 
 // NumPeers returns the number of connected peers.
 func (s *Swarm) NumPeers() int {
+	if s.router == nil {
+		return 0
+	}
+
 	for _, peer := range s.router.Peers.Descriptions() {
 		if peer.Self {
 			return peer.NumConnections

broker/query_test.go

@@ -1 +1,67 @@
 package broker
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/emitter-io/emitter/broker/cluster"
+	"github.com/emitter-io/emitter/broker/message"
+	"github.com/stretchr/testify/assert"
+)
+
+func Test_newQueryManager(t *testing.T) {
+	s := new(Service)
+	q := newQueryManager(s)
+
+	assert.Equal(t, s, q.service)
+	assert.Equal(t, uint32(0), q.next)
+	assert.NotEqual(t, "", q.ID())
+	assert.Equal(t, message.SubscriberDirect, q.Type())
+}
+
+func TestQuerySend_noSSID(t *testing.T) {
+	q := newQueryManager(nil)
+
+	err := q.Send(&message.Message{})
+	assert.Error(t, errors.New("Invalid query received"), err)
+}
+
+func TestQuerySend_Response(t *testing.T) {
+	q := newQueryManager(nil)
+
+	err := q.Send(&message.Message{
+		Channel: []byte("response"),
+		Ssid:    []uint32{0, 1, 2},
+	})
+
+	// There should be no awaiter, hence this should just pass
+	assert.NoError(t, err)
+}
+
+func TestQuerySend_Request(t *testing.T) {
+	q := newQueryManager(&Service{
+		cluster: &cluster.Swarm{},
+	})
+
+	err := q.Send(&message.Message{
+		Channel: []byte("request/12345/"),
+		Ssid:    []uint32{0, 1, 2},
+	})
+
+	assert.Equal(t, "No query handler found for request/12345/", err.Error())
+}
+
+func TestQuery_Query(t *testing.T) {
+	q := newQueryManager(&Service{
+		subscriptions: message.NewTrie(),
+		cluster:       &cluster.Swarm{},
+	})
+
+	awaiter, err := q.Query("test", nil)
+	assert.NoError(t, err)
+	assert.NotNil(t, awaiter)
+
+	result := awaiter.Gather(1 * time.Millisecond)
+	assert.Empty(t, result)
+}