diff --git a/LICENSE b/LICENSE new file mode 100644 index 00000000..8dada3ed --- /dev/null +++ b/LICENSE @@ -0,0 +1,201 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "{}" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright {yyyy} {name of copyright owner} + + 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. diff --git a/Makefile b/Makefile new file mode 100644 index 00000000..295001f8 --- /dev/null +++ b/Makefile @@ -0,0 +1,9 @@ +.PHONY: docs +REPO:=github.com/tendermint/go-events + +docs: + @go get github.com/davecheney/godoc2md + godoc2md $(REPO) > README.md + +test: + go test -v ./... diff --git a/README.md b/README.md new file mode 100644 index 00000000..c435e728 --- /dev/null +++ b/README.md @@ -0,0 +1,175 @@ + + +# events +`import "github.com/tendermint/go-events"` + +* [Overview](#pkg-overview) +* [Index](#pkg-index) + +## Overview +Pub-Sub in go with event caching + + + + +## Index +* [type EventCache](#EventCache) + * [func NewEventCache(evsw Fireable) *EventCache](#NewEventCache) + * [func (evc *EventCache) FireEvent(event string, data EventData)](#EventCache.FireEvent) + * [func (evc *EventCache) Flush()](#EventCache.Flush) +* [type EventCallback](#EventCallback) +* [type EventData](#EventData) +* [type EventSwitch](#EventSwitch) + * [func NewEventSwitch() EventSwitch](#NewEventSwitch) +* [type Eventable](#Eventable) +* [type Fireable](#Fireable) + + +#### Package files +[event_cache.go](/src/github.com/tendermint/go-events/event_cache.go) [events.go](/src/github.com/tendermint/go-events/events.go) [log.go](/src/github.com/tendermint/go-events/log.go) + + + + + + +## type [EventCache](/src/target/event_cache.go?s=152:215#L1) +``` go +type EventCache struct { + // contains filtered or unexported fields +} +``` +An EventCache buffers events for a Fireable +All events are cached. Filtering happens on Flush + + + + + + + +### func [NewEventCache](/src/target/event_cache.go?s=275:320#L5) +``` go +func NewEventCache(evsw Fireable) *EventCache +``` +Create a new EventCache with an EventSwitch as backend + + + + + +### func (\*EventCache) [FireEvent](/src/target/event_cache.go?s=534:596#L19) +``` go +func (evc *EventCache) FireEvent(event string, data EventData) +``` +Cache an event to be fired upon finality. + + + + +### func (\*EventCache) [Flush](/src/target/event_cache.go?s=773:803#L26) +``` go +func (evc *EventCache) Flush() +``` +Fire events by running evsw.FireEvent on all cached events. Blocks. +Clears cached events + + + + +## type [EventCallback](/src/target/events.go?s=4182:4221#L175) +``` go +type EventCallback func(data EventData) +``` + + + + + + + + + +## type [EventData](/src/target/events.go?s=236:287#L4) +``` go +type EventData interface { +} +``` +Generic event data can be typed and registered with tendermint/go-wire +via concrete implementation of this interface + + + + + + + + + + +## type [EventSwitch](/src/target/events.go?s=553:760#L19) +``` go +type EventSwitch interface { + Service + Fireable + + AddListenerForEvent(listenerID, event string, cb EventCallback) + RemoveListenerForEvent(event string, listenerID string) + RemoveListener(listenerID string) +} +``` + + + + + + +### func [NewEventSwitch](/src/target/events.go?s=902:935#L36) +``` go +func NewEventSwitch() EventSwitch +``` + + + + +## type [Eventable](/src/target/events.go?s=371:433#L10) +``` go +type Eventable interface { + SetEventSwitch(evsw EventSwitch) +} +``` +reactors and other modules should export +this interface to become eventable + + + + + + + + + + +## type [Fireable](/src/target/events.go?s=483:551#L15) +``` go +type Fireable interface { + FireEvent(event string, data EventData) +} +``` +an event switch or cache implements fireable + + + + + + + + + + + + + + +- - - +Generated by [godoc2md](http://godoc.org/github.com/davecheney/godoc2md) diff --git a/event_cache.go b/event_cache.go new file mode 100644 index 00000000..905f1096 --- /dev/null +++ b/event_cache.go @@ -0,0 +1,41 @@ +package events + +const ( + eventsBufferSize = 1000 +) + +// An EventCache buffers events for a Fireable +// All events are cached. Filtering happens on Flush +type EventCache struct { + evsw Fireable + events []eventInfo +} + +// Create a new EventCache with an EventSwitch as backend +func NewEventCache(evsw Fireable) *EventCache { + return &EventCache{ + evsw: evsw, + events: make([]eventInfo, eventsBufferSize), + } +} + +// a cached event +type eventInfo struct { + event string + data EventData +} + +// Cache an event to be fired upon finality. +func (evc *EventCache) FireEvent(event string, data EventData) { + // append to list + evc.events = append(evc.events, eventInfo{event, data}) +} + +// Fire events by running evsw.FireEvent on all cached events. Blocks. +// Clears cached events +func (evc *EventCache) Flush() { + for _, ei := range evc.events { + evc.evsw.FireEvent(ei.event, ei.data) + } + evc.events = make([]eventInfo, eventsBufferSize) +} diff --git a/events.go b/events.go new file mode 100644 index 00000000..e613a8e0 --- /dev/null +++ b/events.go @@ -0,0 +1,226 @@ +/* +Pub-Sub in go with event caching +*/ +package events + +import ( + "sync" + + . "github.com/tendermint/go-common" +) + +// Generic event data can be typed and registered with tendermint/go-wire +// via concrete implementation of this interface +type EventData interface { + //AssertIsEventData() +} + +// reactors and other modules should export +// this interface to become eventable +type Eventable interface { + SetEventSwitch(evsw EventSwitch) +} + +// an event switch or cache implements fireable +type Fireable interface { + FireEvent(event string, data EventData) +} + +type EventSwitch interface { + Service + Fireable + + AddListenerForEvent(listenerID, event string, cb EventCallback) + RemoveListenerForEvent(event string, listenerID string) + RemoveListener(listenerID string) +} + +type eventSwitch struct { + BaseService + + mtx sync.RWMutex + eventCells map[string]*eventCell + listeners map[string]*eventListener +} + +func NewEventSwitch() EventSwitch { + evsw := &eventSwitch{} + evsw.BaseService = *NewBaseService(log, "EventSwitch", evsw) + return evsw +} + +func (evsw *eventSwitch) OnStart() error { + evsw.BaseService.OnStart() + evsw.eventCells = make(map[string]*eventCell) + evsw.listeners = make(map[string]*eventListener) + return nil +} + +func (evsw *eventSwitch) OnStop() { + evsw.mtx.Lock() + defer evsw.mtx.Unlock() + evsw.BaseService.OnStop() + evsw.eventCells = nil + evsw.listeners = nil +} + +func (evsw *eventSwitch) AddListenerForEvent(listenerID, event string, cb EventCallback) { + // Get/Create eventCell and listener + evsw.mtx.Lock() + eventCell := evsw.eventCells[event] + if eventCell == nil { + eventCell = newEventCell() + evsw.eventCells[event] = eventCell + } + listener := evsw.listeners[listenerID] + if listener == nil { + listener = newEventListener(listenerID) + evsw.listeners[listenerID] = listener + } + evsw.mtx.Unlock() + + // Add event and listener + eventCell.AddListener(listenerID, cb) + listener.AddEvent(event) +} + +func (evsw *eventSwitch) RemoveListener(listenerID string) { + // Get and remove listener + evsw.mtx.RLock() + listener := evsw.listeners[listenerID] + evsw.mtx.RUnlock() + if listener == nil { + return + } + + evsw.mtx.Lock() + delete(evsw.listeners, listenerID) + evsw.mtx.Unlock() + + // Remove callback for each event. + listener.SetRemoved() + for _, event := range listener.GetEvents() { + evsw.RemoveListenerForEvent(event, listenerID) + } +} + +func (evsw *eventSwitch) RemoveListenerForEvent(event string, listenerID string) { + // Get eventCell + evsw.mtx.Lock() + eventCell := evsw.eventCells[event] + evsw.mtx.Unlock() + + if eventCell == nil { + return + } + + // Remove listenerID from eventCell + numListeners := eventCell.RemoveListener(listenerID) + + // Maybe garbage collect eventCell. + if numListeners == 0 { + // Lock again and double check. + evsw.mtx.Lock() // OUTER LOCK + eventCell.mtx.Lock() // INNER LOCK + if len(eventCell.listeners) == 0 { + delete(evsw.eventCells, event) + } + eventCell.mtx.Unlock() // INNER LOCK + evsw.mtx.Unlock() // OUTER LOCK + } +} + +func (evsw *eventSwitch) FireEvent(event string, data EventData) { + // Get the eventCell + evsw.mtx.RLock() + eventCell := evsw.eventCells[event] + evsw.mtx.RUnlock() + + if eventCell == nil { + return + } + + // Fire event for all listeners in eventCell + eventCell.FireEvent(data) +} + +//----------------------------------------------------------------------------- + +// eventCell handles keeping track of listener callbacks for a given event. +type eventCell struct { + mtx sync.RWMutex + listeners map[string]EventCallback +} + +func newEventCell() *eventCell { + return &eventCell{ + listeners: make(map[string]EventCallback), + } +} + +func (cell *eventCell) AddListener(listenerID string, cb EventCallback) { + cell.mtx.Lock() + cell.listeners[listenerID] = cb + cell.mtx.Unlock() +} + +func (cell *eventCell) RemoveListener(listenerID string) int { + cell.mtx.Lock() + delete(cell.listeners, listenerID) + numListeners := len(cell.listeners) + cell.mtx.Unlock() + return numListeners +} + +func (cell *eventCell) FireEvent(data EventData) { + cell.mtx.RLock() + for _, listener := range cell.listeners { + listener(data) + } + cell.mtx.RUnlock() +} + +//----------------------------------------------------------------------------- + +type EventCallback func(data EventData) + +type eventListener struct { + id string + + mtx sync.RWMutex + removed bool + events []string +} + +func newEventListener(id string) *eventListener { + return &eventListener{ + id: id, + removed: false, + events: nil, + } +} + +func (evl *eventListener) AddEvent(event string) { + evl.mtx.Lock() + defer evl.mtx.Unlock() + + if evl.removed { + return + } + evl.events = append(evl.events, event) +} + +func (evl *eventListener) GetEvents() []string { + evl.mtx.RLock() + defer evl.mtx.RUnlock() + + events := make([]string, len(evl.events)) + copy(events, evl.events) + return events +} + +func (evl *eventListener) SetRemoved() { + evl.mtx.Lock() + defer evl.mtx.Unlock() + evl.removed = true +} diff --git a/events_test.go b/events_test.go new file mode 100644 index 00000000..c1b48b16 --- /dev/null +++ b/events_test.go @@ -0,0 +1,381 @@ +package events + +import ( + "fmt" + "math/rand" + "testing" + "time" + + "github.com/stretchr/testify/assert" +) + +// TestAddListenerForEventFireOnce sets up an EventSwitch, subscribes a single +// listener to an event, and sends a string "data". +func TestAddListenerForEventFireOnce(t *testing.T) { + evsw := NewEventSwitch() + started, err := evsw.Start() + if started == false || err != nil { + t.Errorf("Failed to start EventSwitch, error: %v", err) + } + messages := make(chan EventData) + evsw.AddListenerForEvent("listener", "event", + func(data EventData) { + messages <- data + }) + go evsw.FireEvent("event", "data") + received := <-messages + if received != "data" { + t.Errorf("Message received does not match: %v", received) + } +} + +// TestAddListenerForEventFireMany sets up an EventSwitch, subscribes a single +// listener to an event, and sends a thousand integers. +func TestAddListenerForEventFireMany(t *testing.T) { + evsw := NewEventSwitch() + started, err := evsw.Start() + if started == false || err != nil { + t.Errorf("Failed to start EventSwitch, error: %v", err) + } + doneSum := make(chan uint64) + doneSending := make(chan uint64) + numbers := make(chan uint64, 4) + // subscribe one listener for one event + evsw.AddListenerForEvent("listener", "event", + func(data EventData) { + numbers <- data.(uint64) + }) + // collect received events + go sumReceivedNumbers(numbers, doneSum) + // go fire events + go fireEvents(evsw, "event", doneSending, uint64(1)) + checkSum := <-doneSending + close(numbers) + eventSum := <-doneSum + if checkSum != eventSum { + t.Errorf("Not all messages sent were received.\n") + } +} + +// TestAddListenerForDifferentEvents sets up an EventSwitch, subscribes a single +// listener to three different events and sends a thousand integers for each +// of the three events. +func TestAddListenerForDifferentEvents(t *testing.T) { + evsw := NewEventSwitch() + started, err := evsw.Start() + if started == false || err != nil { + t.Errorf("Failed to start EventSwitch, error: %v", err) + } + doneSum := make(chan uint64) + doneSending1 := make(chan uint64) + doneSending2 := make(chan uint64) + doneSending3 := make(chan uint64) + numbers := make(chan uint64, 4) + // subscribe one listener to three events + evsw.AddListenerForEvent("listener", "event1", + func(data EventData) { + numbers <- data.(uint64) + }) + evsw.AddListenerForEvent("listener", "event2", + func(data EventData) { + numbers <- data.(uint64) + }) + evsw.AddListenerForEvent("listener", "event3", + func(data EventData) { + numbers <- data.(uint64) + }) + // collect received events + go sumReceivedNumbers(numbers, doneSum) + // go fire events + go fireEvents(evsw, "event1", doneSending1, uint64(1)) + go fireEvents(evsw, "event2", doneSending2, uint64(1)) + go fireEvents(evsw, "event3", doneSending3, uint64(1)) + var checkSum uint64 = 0 + checkSum += <-doneSending1 + checkSum += <-doneSending2 + checkSum += <-doneSending3 + close(numbers) + eventSum := <-doneSum + if checkSum != eventSum { + t.Errorf("Not all messages sent were received.\n") + } +} + +// TestAddDifferentListenerForDifferentEvents sets up an EventSwitch, +// subscribes a first listener to three events, and subscribes a second +// listener to two of those three events, and then sends a thousand integers +// for each of the three events. +func TestAddDifferentListenerForDifferentEvents(t *testing.T) { + evsw := NewEventSwitch() + started, err := evsw.Start() + if started == false || err != nil { + t.Errorf("Failed to start EventSwitch, error: %v", err) + } + doneSum1 := make(chan uint64) + doneSum2 := make(chan uint64) + doneSending1 := make(chan uint64) + doneSending2 := make(chan uint64) + doneSending3 := make(chan uint64) + numbers1 := make(chan uint64, 4) + numbers2 := make(chan uint64, 4) + // subscribe two listener to three events + evsw.AddListenerForEvent("listener1", "event1", + func(data EventData) { + numbers1 <- data.(uint64) + }) + evsw.AddListenerForEvent("listener1", "event2", + func(data EventData) { + numbers1 <- data.(uint64) + }) + evsw.AddListenerForEvent("listener1", "event3", + func(data EventData) { + numbers1 <- data.(uint64) + }) + evsw.AddListenerForEvent("listener2", "event2", + func(data EventData) { + numbers2 <- data.(uint64) + }) + evsw.AddListenerForEvent("listener2", "event3", + func(data EventData) { + numbers2 <- data.(uint64) + }) + // collect received events for listener1 + go sumReceivedNumbers(numbers1, doneSum1) + // collect received events for listener2 + go sumReceivedNumbers(numbers2, doneSum2) + // go fire events + go fireEvents(evsw, "event1", doneSending1, uint64(1)) + go fireEvents(evsw, "event2", doneSending2, uint64(1001)) + go fireEvents(evsw, "event3", doneSending3, uint64(2001)) + checkSumEvent1 := <-doneSending1 + checkSumEvent2 := <-doneSending2 + checkSumEvent3 := <-doneSending3 + checkSum1 := checkSumEvent1 + checkSumEvent2 + checkSumEvent3 + checkSum2 := checkSumEvent2 + checkSumEvent3 + close(numbers1) + close(numbers2) + eventSum1 := <-doneSum1 + eventSum2 := <-doneSum2 + if checkSum1 != eventSum1 || + checkSum2 != eventSum2 { + t.Errorf("Not all messages sent were received for different listeners to different events.\n") + } +} + +// TestAddAndRemoveListener sets up an EventSwitch, subscribes a listener to +// two events, fires a thousand integers for the first event, then unsubscribes +// the listener and fires a thousand integers for the second event. +func TestAddAndRemoveListener(t *testing.T) { + evsw := NewEventSwitch() + started, err := evsw.Start() + if started == false || err != nil { + t.Errorf("Failed to start EventSwitch, error: %v", err) + } + doneSum1 := make(chan uint64) + doneSum2 := make(chan uint64) + doneSending1 := make(chan uint64) + doneSending2 := make(chan uint64) + numbers1 := make(chan uint64, 4) + numbers2 := make(chan uint64, 4) + // subscribe two listener to three events + evsw.AddListenerForEvent("listener", "event1", + func(data EventData) { + numbers1 <- data.(uint64) + }) + evsw.AddListenerForEvent("listener", "event2", + func(data EventData) { + numbers2 <- data.(uint64) + }) + // collect received events for event1 + go sumReceivedNumbers(numbers1, doneSum1) + // collect received events for event2 + go sumReceivedNumbers(numbers2, doneSum2) + // go fire events + go fireEvents(evsw, "event1", doneSending1, uint64(1)) + checkSumEvent1 := <-doneSending1 + // after sending all event1, unsubscribe for all events + evsw.RemoveListener("listener") + go fireEvents(evsw, "event2", doneSending2, uint64(1001)) + checkSumEvent2 := <-doneSending2 + close(numbers1) + close(numbers2) + eventSum1 := <-doneSum1 + eventSum2 := <-doneSum2 + if checkSumEvent1 != eventSum1 || + // correct value asserted by preceding tests, suffices to be non-zero + checkSumEvent2 == uint64(0) || + eventSum2 != uint64(0) { + t.Errorf("Not all messages sent were received or unsubscription did not register.\n") + } +} + +// TestRemoveListener does basic tests on adding and removing +func TestRemoveListener(t *testing.T) { + evsw := NewEventSwitch() + started, err := evsw.Start() + if started == false || err != nil { + t.Errorf("Failed to start EventSwitch, error: %v", err) + } + count := 10 + sum1, sum2 := 0, 0 + // add some listeners and make sure they work + evsw.AddListenerForEvent("listener", "event1", + func(data EventData) { + sum1 += 1 + }) + evsw.AddListenerForEvent("listener", "event2", + func(data EventData) { + sum2 += 1 + }) + for i := 0; i < count; i++ { + evsw.FireEvent("event1", true) + evsw.FireEvent("event2", true) + } + assert.Equal(t, count, sum1) + assert.Equal(t, count, sum2) + + // remove one by event and make sure it is gone + evsw.RemoveListenerForEvent("event2", "listener") + for i := 0; i < count; i++ { + evsw.FireEvent("event1", true) + evsw.FireEvent("event2", true) + } + assert.Equal(t, count*2, sum1) + assert.Equal(t, count, sum2) + + // remove the listener entirely and make sure both gone + evsw.RemoveListener("listener") + for i := 0; i < count; i++ { + evsw.FireEvent("event1", true) + evsw.FireEvent("event2", true) + } + assert.Equal(t, count*2, sum1) + assert.Equal(t, count, sum2) +} + +// TestAddAndRemoveListenersAsync sets up an EventSwitch, subscribes two +// listeners to three events, and fires a thousand integers for each event. +// These two listeners serve as the baseline validation while other listeners +// are randomly subscribed and unsubscribed. +// More precisely it randomly subscribes new listeners (different from the first +// two listeners) to one of these three events. At the same time it starts +// randomly unsubscribing these additional listeners from all events they are +// at that point subscribed to. +// NOTE: it is important to run this test with race conditions tracking on, +// `go test -race`, to examine for possible race conditions. +func TestRemoveListenersAsync(t *testing.T) { + evsw := NewEventSwitch() + started, err := evsw.Start() + if started == false || err != nil { + t.Errorf("Failed to start EventSwitch, error: %v", err) + } + doneSum1 := make(chan uint64) + doneSum2 := make(chan uint64) + doneSending1 := make(chan uint64) + doneSending2 := make(chan uint64) + doneSending3 := make(chan uint64) + numbers1 := make(chan uint64, 4) + numbers2 := make(chan uint64, 4) + // subscribe two listener to three events + evsw.AddListenerForEvent("listener1", "event1", + func(data EventData) { + numbers1 <- data.(uint64) + }) + evsw.AddListenerForEvent("listener1", "event2", + func(data EventData) { + numbers1 <- data.(uint64) + }) + evsw.AddListenerForEvent("listener1", "event3", + func(data EventData) { + numbers1 <- data.(uint64) + }) + evsw.AddListenerForEvent("listener2", "event1", + func(data EventData) { + numbers2 <- data.(uint64) + }) + evsw.AddListenerForEvent("listener2", "event2", + func(data EventData) { + numbers2 <- data.(uint64) + }) + evsw.AddListenerForEvent("listener2", "event3", + func(data EventData) { + numbers2 <- data.(uint64) + }) + // collect received events for event1 + go sumReceivedNumbers(numbers1, doneSum1) + // collect received events for event2 + go sumReceivedNumbers(numbers2, doneSum2) + addListenersStress := func() { + s1 := rand.NewSource(time.Now().UnixNano()) + r1 := rand.New(s1) + for k := uint16(0); k < 400; k++ { + listenerNumber := r1.Intn(100) + 3 + eventNumber := r1.Intn(3) + 1 + go evsw.AddListenerForEvent(fmt.Sprintf("listener%v", listenerNumber), + fmt.Sprintf("event%v", eventNumber), + func(_ EventData) {}) + } + } + removeListenersStress := func() { + s2 := rand.NewSource(time.Now().UnixNano()) + r2 := rand.New(s2) + for k := uint16(0); k < 80; k++ { + listenerNumber := r2.Intn(100) + 3 + go evsw.RemoveListener(fmt.Sprintf("listener%v", listenerNumber)) + } + } + addListenersStress() + // go fire events + go fireEvents(evsw, "event1", doneSending1, uint64(1)) + removeListenersStress() + go fireEvents(evsw, "event2", doneSending2, uint64(1001)) + go fireEvents(evsw, "event3", doneSending3, uint64(2001)) + checkSumEvent1 := <-doneSending1 + checkSumEvent2 := <-doneSending2 + checkSumEvent3 := <-doneSending3 + checkSum := checkSumEvent1 + checkSumEvent2 + checkSumEvent3 + close(numbers1) + close(numbers2) + eventSum1 := <-doneSum1 + eventSum2 := <-doneSum2 + if checkSum != eventSum1 || + checkSum != eventSum2 { + t.Errorf("Not all messages sent were received.\n") + } +} + +//------------------------------------------------------------------------------ +// Helper functions + +// sumReceivedNumbers takes two channels and adds all numbers received +// until the receiving channel `numbers` is closed; it then sends the sum +// on `doneSum` and closes that channel. Expected to be run in a go-routine. +func sumReceivedNumbers(numbers, doneSum chan uint64) { + var sum uint64 = 0 + for { + j, more := <-numbers + sum += j + if !more { + doneSum <- sum + close(doneSum) + return + } + } +} + +// fireEvents takes an EventSwitch and fires a thousand integers under +// a given `event` with the integers mootonically increasing from `offset` +// to `offset` + 999. It additionally returns the addition of all integers +// sent on `doneChan` for assertion that all events have been sent, and enabling +// the test to assert all events have also been received. +func fireEvents(evsw EventSwitch, event string, doneChan chan uint64, + offset uint64) { + var sentSum uint64 = 0 + for i := offset; i <= offset+uint64(999); i++ { + sentSum += i + evsw.FireEvent(event, i) + } + doneChan <- sentSum + close(doneChan) + return +} diff --git a/log.go b/log.go new file mode 100644 index 00000000..52546229 --- /dev/null +++ b/log.go @@ -0,0 +1,7 @@ +package events + +import ( + "github.com/tendermint/go-logger" +) + +var log = logger.New("module", "events")