pubsub.in-memory: add support for wildcard topics

Fixes dapr#1964

Signed-off-by: ItalyPaleAle <[email protected]>

go.mod

@@ -37,7 +37,6 @@ require (
 	github.com/andybalholm/brotli v1.0.2 // indirect
 	github.com/apache/rocketmq-client-go/v2 v2.1.1-rc2
 	github.com/apache/thrift v0.16.0 // indirect
-	github.com/asaskevich/EventBus v0.0.0-20200907212545-49d423059eef
 	github.com/aws/aws-sdk-go v1.43.16
 	github.com/baiyubin/aliyun-sts-go-sdk v0.0.0-20180326062324-cfa1a18b161f // indirect
 	github.com/bradfitz/gomemcache v0.0.0-20190913173617-a41fca850d0b

go.sum

@@ -369,8 +369,6 @@ github.com/armon/go-metrics v0.3.10/go.mod h1:4O98XIr/9W0sxpJ8UaYkvjk10Iff7SnFrb
 github.com/armon/go-radix v0.0.0-20180808171621-7fddfc383310/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/armon/go-radix v1.0.0/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgIH9cCH8=
 github.com/aryann/difflib v0.0.0-20170710044230-e206f873d14a/go.mod h1:DAHtR1m6lCRdSC2Tm3DSWRPvIPr6xNKyeHdqDQSQT+A=
-github.com/asaskevich/EventBus v0.0.0-20200907212545-49d423059eef h1:2JGTg6JapxP9/R33ZaagQtAM4EkkSYnIAlOG5EI8gkM=
-github.com/asaskevich/EventBus v0.0.0-20200907212545-49d423059eef/go.mod h1:JS7hed4L1fj0hXcyEejnW57/7LCetXggd+vwrRnYeII=
 github.com/asaskevich/govalidator v0.0.0-20190424111038-f61b66f89f4a/go.mod h1:lB+ZfQJz7igIIfQNfa7Ml4HSf2uFQQRzpGGRXenZAgY=
 github.com/asaskevich/govalidator v0.0.0-20200108200545-475eaeb16496 h1:zV3ejI06GQ59hwDQAvmK1qxOQGB3WuVTRoY0okPTAv0=
 github.com/asaskevich/govalidator v0.0.0-20200108200545-475eaeb16496/go.mod h1:oGkLhpf+kjZl6xBf758TQhh5XrAeiJv/7FRz/2spLIg=

internal/eventbus/README.md

@@ -0,0 +1,29 @@
+# eventbus
+
+This is a based on [github.com/asaskevich/EventBus](https://github.com/asaskevich/EventBus) (commit: 49d423059eefd67a7243331db83e16d347139b4a), picking only the in-memory event bus and with modifications to add support for wildcard topics.
+
+License for the original code:
+
+```
+The MIT License (MIT)
+
+Copyright (c) 2014 Alex Saskevich
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+```

internal/eventbus/event_bus.go

@@ -0,0 +1,223 @@
+/*
+Copyright 2021 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+/*
+The MIT License (MIT)
+Copyright (c) 2014 Alex Saskevich
+*/
+
+package eventbus
+
+import (
+	"fmt"
+	"reflect"
+	"strings"
+	"sync"
+)
+
+// BusSubscriber defines subscription-related bus behavior
+type BusSubscriber interface {
+	Subscribe(topic string, fn interface{}) error
+	SubscribeAsync(topic string, fn interface{}, transactional bool) error
+	Unsubscribe(topic string, handler interface{}) error
+	WaitAsync()
+}
+
+// BusPublisher defines publishing-related bus behavior
+type BusPublisher interface {
+	Publish(topic string, args ...interface{})
+}
+
+// Bus englobes global (subscribe, publish, control) bus behavior
+type Bus interface {
+	BusSubscriber
+	BusPublisher
+}
+
+// EventBus - box for handlers and callbacks.
+type EventBus struct {
+	enableWildcards bool
+	handlers        map[string][]*eventHandler
+	lock            sync.Mutex // a lock for the map
+	wg              sync.WaitGroup
+}
+
+type eventHandler struct {
+	callBack      reflect.Value
+	async         bool
+	transactional bool
+	sync.Mutex    // lock for an event handler - useful for running async callbacks serially
+}
+
+// New returns new EventBus with empty handlers.
+func New(enableWildcards bool) Bus {
+	b := &EventBus{
+		enableWildcards,
+		make(map[string][]*eventHandler),
+		sync.Mutex{},
+		sync.WaitGroup{},
+	}
+	return Bus(b)
+}
+
+// doSubscribe handles the subscription logic and is utilized by the public Subscribe functions
+func (bus *EventBus) doSubscribe(topic string, fn interface{}, handler *eventHandler) error {
+	bus.lock.Lock()
+	defer bus.lock.Unlock()
+	if !(reflect.TypeOf(fn).Kind() == reflect.Func) {
+		return fmt.Errorf("%s is not of type reflect.Func", reflect.TypeOf(fn).Kind())
+	}
+	bus.handlers[topic] = append(bus.handlers[topic], handler)
+	return nil
+}
+
+// Subscribe subscribes to a topic.
+// Returns error if `fn` is not a function.
+func (bus *EventBus) Subscribe(topic string, fn interface{}) error {
+	return bus.doSubscribe(topic, fn, &eventHandler{
+		reflect.ValueOf(fn), false, false, sync.Mutex{},
+	})
+}
+
+// SubscribeAsync subscribes to a topic with an asynchronous callback
+// Transactional determines whether subsequent callbacks for a topic are
+// run serially (true) or concurrently (false)
+// Returns error if `fn` is not a function.
+func (bus *EventBus) SubscribeAsync(topic string, fn interface{}, transactional bool) error {
+	return bus.doSubscribe(topic, fn, &eventHandler{
+		reflect.ValueOf(fn), true, transactional, sync.Mutex{},
+	})
+}
+
+// getCallbacks returns the callback(s) registered for the topic
+func (bus *EventBus) getCallbacks(topic string) []*eventHandler {
+	if !bus.enableWildcards {
+		handlers, ok := bus.handlers[topic]
+		if ok && len(handlers) > 0 {
+			// Make a hard copy to prevent the slice to be changed after this method
+			copyHandlers := make([]*eventHandler, len(handlers))
+			copy(copyHandlers, handlers)
+			return copyHandlers
+		}
+		return nil
+	}
+
+	handlers := []*eventHandler{}
+	for k, h := range bus.handlers {
+		if k == topic ||
+			(strings.HasSuffix(k, "*") && strings.HasPrefix(topic, k[0:len(k)-1]) && topic != k[0:len(k)-1]) {
+			handlers = append(handlers, h...)
+			continue
+		}
+	}
+
+	if len(handlers) == 0 {
+		return nil
+	}
+	return handlers
+}
+
+// Unsubscribe removes callback defined for a topic.
+// Returns error if there are no callbacks subscribed to the topic.
+func (bus *EventBus) Unsubscribe(topic string, handler interface{}) error {
+	bus.lock.Lock()
+	defer bus.lock.Unlock()
+
+	if _, ok := bus.handlers[topic]; ok && len(bus.handlers[topic]) > 0 {
+		bus.removeHandler(topic, bus.findHandlerIdx(topic, reflect.ValueOf(handler)))
+		return nil
+	}
+	return fmt.Errorf("topic %s doesn't exist", topic)
+}
+
+// Publish executes callback defined for a topic. Any additional argument will be transferred to the callback.
+func (bus *EventBus) Publish(topic string, args ...interface{}) {
+	bus.lock.Lock() // will unlock if handler is not found or always after setUpPublish
+	defer bus.lock.Unlock()
+
+	handlers := bus.getCallbacks(topic)
+	if len(handlers) > 0 {
+		for _, handler := range handlers {
+			if !handler.async {
+				bus.doPublish(handler, topic, args...)
+			} else {
+				bus.wg.Add(1)
+				if handler.transactional {
+					bus.lock.Unlock()
+					handler.Lock()
+					bus.lock.Lock()
+				}
+				go bus.doPublishAsync(handler, topic, args...)
+			}
+		}
+	}
+}
+
+func (bus *EventBus) doPublish(handler *eventHandler, topic string, args ...interface{}) {
+	passedArguments := bus.setUpPublish(handler, args...)
+	handler.callBack.Call(passedArguments)
+}
+
+func (bus *EventBus) doPublishAsync(handler *eventHandler, topic string, args ...interface{}) {
+	defer bus.wg.Done()
+	if handler.transactional {
+		defer handler.Unlock()
+	}
+	bus.doPublish(handler, topic, args...)
+}
+
+func (bus *EventBus) removeHandler(topic string, idx int) {
+	if _, ok := bus.handlers[topic]; !ok {
+		return
+	}
+	l := len(bus.handlers[topic])
+
+	if !(0 <= idx && idx < l) {
+		return
+	}
+
+	copy(bus.handlers[topic][idx:], bus.handlers[topic][idx+1:])
+	bus.handlers[topic][l-1] = nil // or the zero value of T
+	bus.handlers[topic] = bus.handlers[topic][:l-1]
+}
+
+func (bus *EventBus) findHandlerIdx(topic string, callback reflect.Value) int {
+	if _, ok := bus.handlers[topic]; ok {
+		for idx, handler := range bus.handlers[topic] {
+			if handler.callBack.Type() == callback.Type() &&
+				handler.callBack.Pointer() == callback.Pointer() {
+				return idx
+			}
+		}
+	}
+	return -1
+}
+
+func (bus *EventBus) setUpPublish(callback *eventHandler, args ...interface{}) []reflect.Value {
+	funcType := callback.callBack.Type()
+	passedArguments := make([]reflect.Value, len(args))
+	for i, v := range args {
+		if v == nil {
+			passedArguments[i] = reflect.New(funcType.In(i)).Elem()
+		} else {
+			passedArguments[i] = reflect.ValueOf(v)
+		}
+	}
+
+	return passedArguments
+}
+
+// WaitAsync waits for all async callbacks to complete
+func (bus *EventBus) WaitAsync() {
+	bus.wg.Wait()
+}