diff --git a/pkg/broker/client/sse_test.go b/pkg/broker/client/sse_test.go index 0ee62c47c..b55c9abca 100644 --- a/pkg/broker/client/sse_test.go +++ b/pkg/broker/client/sse_test.go @@ -193,7 +193,6 @@ func TestBrokerMultiSubscriberCustomObject(t *testing.T) { go func() { for { <-time.After(10 * time.Millisecond) - now := time.Now() evt := event{Time: now.UnixNano(), Message: fmt.Sprintf("Now is %v", now)} require.NoError(t, broker.Publish("remote/instance1", evt)) @@ -214,6 +213,86 @@ func TestBrokerMultiSubscriberCustomObject(t *testing.T) { } +func TestBrokerMultiSubscriberPartialMatchTopic(t *testing.T) { + type event struct { + Time int64 + Message string + } + + socketFile := tempSocket() + socket := "unix://broker" + socketFile + + broker, err := server.ListenAndServeOnSocket(socketFile) + require.NoError(t, err) + + received1 := make(chan event) + received2 := make(chan event) + + opts := Options{SocketDir: filepath.Dir(socketFile)} + + topic1, errs1, err := Subscribe(socket, "local/instance", opts) + require.NoError(t, err) + go func() { + for { + select { + case e := <-errs1: + panic(e) + case m, ok := <-topic1: + if ok { + var val event + require.NoError(t, m.Decode(&val)) + received1 <- val + } else { + close(received1) + } + } + } + }() + + topic2, errs2, err := Subscribe(socket, "local/instancetest", opts) + require.NoError(t, err) + go func() { + for { + select { + case e := <-errs2: + panic(e) + case m, ok := <-topic2: + if ok { + var val event + require.NoError(t, m.Decode(&val)) + received2 <- val + } else { + close(received2) + } + } + } + }() + + go func() { + for { + <-time.After(10 * time.Millisecond) + now := time.Now() + evt := event{Time: now.UnixNano(), Message: fmt.Sprintf("Now is %v", now)} + require.NoError(t, broker.Publish("local/instance", evt)) + evt = event{Time: now.Add(1 * time.Minute).UnixNano(), Message: fmt.Sprintf("Now is %v", now.Add(1*time.Minute))} + require.NoError(t, broker.Publish("local/instancetest", evt)) + } + }() + + // Test a few rounds to make sure all subscribers get the same messages each round. + for i := 0; i < 5; i++ { + fmt.Print("runtestcycle") + b := <-received2 + a := <-received1 + require.NotNil(t, a) + require.NotEqual(t, "", a.Message) + require.NotEqual(t, a, b) + } + + broker.Stop() + +} + // This tests the case where the broker is mapped to url route (e.g. /events) // In this case, we need to specify the path option in the connection options so that // we can properly connect to the broker at the url prefix. diff --git a/pkg/broker/server/sse.go b/pkg/broker/server/sse.go index 298d4270b..f368108ef 100644 --- a/pkg/broker/server/sse.go +++ b/pkg/broker/server/sse.go @@ -4,6 +4,7 @@ import ( "bytes" "fmt" "net/http" + "strings" "time" log "github.com/Sirupsen/logrus" @@ -159,6 +160,19 @@ func (b *Broker) ServeHTTP(rw http.ResponseWriter, req *http.Request) { } } +//checkPath Compare the subscribed topic with the published topic to see if the message can be sent to the client. +func (b *Broker) checkPath(subscribedPath string, publishedPath string) bool { + if subscribedPath != "/" { + pPath := strings.Split(publishedPath, "/") + for i, sliceTopic := range strings.Split(subscribedPath, "/") { + if pPath[i] != sliceTopic { + return false + } + } + } + return true +} + func (b *Broker) run() { for { select { @@ -244,6 +258,10 @@ func (b *Broker) run() { panic("assert-failed") } + // Make sure that the topic subscribed to by the client is the upper topic of the topic being notified or the topic exactly matched. + if !b.checkPath(key, event.topic) { + return false + } for ch := range chset { select {