Indexing error due to multiple removeHandler

[Jjjpan]

In line 141 func Publish

// 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()
    if handlers, ok := bus.handlers[topic]; ok && 0 < len(handlers) {
        // Handlers slice may be changed by removeHandler and Unsubscribe during iteration,
        // so make a copy and iterate the copied slice.
        copyHandlers := make([]*eventHandler, len(handlers))
        copy(copyHandlers, handlers)
        for i, handler := range copyHandlers {
            if handler.flagOnce {
                bus.removeHandler(topic, i) // multiple operation causes indexing error
            }
            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...)
            }
        }
    }
}

i and handler are ranged in copyHandlers while remove operation actions in bus.handlers. 🤨

[Jjjpan]

In a test case of many subscribe and many subscribeOnce, bus.removeHandler tries to remove a handler with index out of range.

Add out-of-range outputs in event_bus.go:

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) {
                fmt.Println("out of range")
        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]
}

Test case in event_bus_test.go:

func TestManySubscribeAndManySubscribe(t *testing.T) {
    bus := New()
    event := "topic"
    flag := 0
    fn := func() { flag += 1 }
    bus.SubscribeOnce(event, fn)
    bus.SubscribeOnce(event, fn)
    bus.Subscribe(event, fn)
    bus.Subscribe(event, fn)
    bus.SubscribeOnce(event, fn)
    bus.SubscribeOnce(event, fn)
    bus.Publish(event)

    if flag != 6 {
        t.Fail()
    }
}

Test result:

=== RUN   TestManySubscribeAndManySubscribe
remove out of range
remove out of range
--- PASS: TestManySubscribeAndManySubscribe (0.00s)
PASS
ok      eventbus/v1     (cached)

[Jjjpan]

Would it be better to use offset to record the offset of index between handlers and copyHandlers.🙄

Maybe like this:

func (bus *EventBus) Publish(topic string, args ...interface{}) {
    bus.lock.Lock()
    defer bus.lock.Unlock()

    if topicHandlers, ok := bus.handlers[topic]; ok && len(topicHandlers) > 0 {
        copiedHandlers := make([]*eventHandler, len(topicHandlers))
        copy(copiedHandlers, topicHandlers)
        offset := 0
        for index, handler := range copiedHandlers {
            if handler.flagOnce {
                bus.removeHandler(topic, index+offset)
                offset--
            }
            if !handler.async {
                bus.doPublish(handler, topic, args...)
            } else {
                bus.wg.Add(1)
                if handler.transactional {
                    bus.lock.Unlock()
                    handler.lock.Lock()
                    bus.lock.Lock()
                }
                go bus.doPublishAsync(handler, topic, args...)
            }
        }
    }
}

[varfrog]

This project seems to not be maintained any more. I'm just looking around the issues here. Maybe there's a maintained fork...