diff --git a/event/event.go b/event/event.go
deleted file mode 100644
index 25a2c2e457..0000000000
--- a/event/event.go
+++ /dev/null
@@ -1,217 +0,0 @@
-// Copyright 2014 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-// Package event deals with subscriptions to real-time events.
-package event
-
-import (
- "errors"
- "fmt"
- "reflect"
- "sync"
- "time"
-)
-
-// TypeMuxEvent is a time-tagged notification pushed to subscribers.
-type TypeMuxEvent struct {
- Time time.Time
- Data interface{}
-}
-
-// A TypeMux dispatches events to registered receivers. Receivers can be
-// registered to handle events of certain type. Any operation
-// called after mux is stopped will return ErrMuxClosed.
-//
-// The zero value is ready to use.
-//
-// Deprecated: use Feed
-type TypeMux struct {
- mutex sync.RWMutex
- subm map[reflect.Type][]*TypeMuxSubscription
- stopped bool
-}
-
-// ErrMuxClosed is returned when Posting on a closed TypeMux.
-var ErrMuxClosed = errors.New("event: mux closed")
-
-// Subscribe creates a subscription for events of the given types. The
-// subscription's channel is closed when it is unsubscribed
-// or the mux is closed.
-func (mux *TypeMux) Subscribe(types ...interface{}) *TypeMuxSubscription {
- sub := newsub(mux)
- mux.mutex.Lock()
- defer mux.mutex.Unlock()
- if mux.stopped {
- // set the status to closed so that calling Unsubscribe after this
- // call will short circuit.
- sub.closed = true
- close(sub.postC)
- } else {
- if mux.subm == nil {
- mux.subm = make(map[reflect.Type][]*TypeMuxSubscription, len(types))
- }
- for _, t := range types {
- rtyp := reflect.TypeOf(t)
- oldsubs := mux.subm[rtyp]
- if find(oldsubs, sub) != -1 {
- panic(fmt.Sprintf("event: duplicate type %s in Subscribe", rtyp))
- }
- subs := make([]*TypeMuxSubscription, len(oldsubs)+1)
- copy(subs, oldsubs)
- subs[len(oldsubs)] = sub
- mux.subm[rtyp] = subs
- }
- }
- return sub
-}
-
-// Post sends an event to all receivers registered for the given type.
-// It returns ErrMuxClosed if the mux has been stopped.
-func (mux *TypeMux) Post(ev interface{}) error {
- event := &TypeMuxEvent{
- Time: time.Now(),
- Data: ev,
- }
- rtyp := reflect.TypeOf(ev)
- mux.mutex.RLock()
- if mux.stopped {
- mux.mutex.RUnlock()
- return ErrMuxClosed
- }
- subs := mux.subm[rtyp]
- mux.mutex.RUnlock()
- for _, sub := range subs {
- sub.deliver(event)
- }
- return nil
-}
-
-// Stop closes a mux. The mux can no longer be used.
-// Future Post calls will fail with ErrMuxClosed.
-// Stop blocks until all current deliveries have finished.
-func (mux *TypeMux) Stop() {
- mux.mutex.Lock()
- defer mux.mutex.Unlock()
- for _, subs := range mux.subm {
- for _, sub := range subs {
- sub.closewait()
- }
- }
- mux.subm = nil
- mux.stopped = true
-}
-
-func (mux *TypeMux) del(s *TypeMuxSubscription) {
- mux.mutex.Lock()
- defer mux.mutex.Unlock()
- for typ, subs := range mux.subm {
- if pos := find(subs, s); pos >= 0 {
- if len(subs) == 1 {
- delete(mux.subm, typ)
- } else {
- mux.subm[typ] = posdelete(subs, pos)
- }
- }
- }
-}
-
-func find(slice []*TypeMuxSubscription, item *TypeMuxSubscription) int {
- for i, v := range slice {
- if v == item {
- return i
- }
- }
- return -1
-}
-
-func posdelete(slice []*TypeMuxSubscription, pos int) []*TypeMuxSubscription {
- news := make([]*TypeMuxSubscription, len(slice)-1)
- copy(news[:pos], slice[:pos])
- copy(news[pos:], slice[pos+1:])
- return news
-}
-
-// TypeMuxSubscription is a subscription established through TypeMux.
-type TypeMuxSubscription struct {
- mux *TypeMux
- created time.Time
- closeMu sync.Mutex
- closing chan struct{}
- closed bool
-
- // these two are the same channel. they are stored separately so
- // postC can be set to nil without affecting the return value of
- // Chan.
- postMu sync.RWMutex
- readC <-chan *TypeMuxEvent
- postC chan<- *TypeMuxEvent
-}
-
-func newsub(mux *TypeMux) *TypeMuxSubscription {
- c := make(chan *TypeMuxEvent)
- return &TypeMuxSubscription{
- mux: mux,
- created: time.Now(),
- readC: c,
- postC: c,
- closing: make(chan struct{}),
- }
-}
-
-func (s *TypeMuxSubscription) Chan() <-chan *TypeMuxEvent {
- return s.readC
-}
-
-func (s *TypeMuxSubscription) Unsubscribe() {
- s.mux.del(s)
- s.closewait()
-}
-
-func (s *TypeMuxSubscription) Closed() bool {
- s.closeMu.Lock()
- defer s.closeMu.Unlock()
- return s.closed
-}
-
-func (s *TypeMuxSubscription) closewait() {
- s.closeMu.Lock()
- defer s.closeMu.Unlock()
- if s.closed {
- return
- }
- close(s.closing)
- s.closed = true
-
- s.postMu.Lock()
- defer s.postMu.Unlock()
- close(s.postC)
- s.postC = nil
-}
-
-func (s *TypeMuxSubscription) deliver(event *TypeMuxEvent) {
- // Short circuit delivery if stale event
- if s.created.After(event.Time) {
- return
- }
- // Otherwise deliver the event
- s.postMu.RLock()
- defer s.postMu.RUnlock()
-
- select {
- case s.postC <- event:
- case <-s.closing:
- }
-}
diff --git a/event/event_test.go b/event/event_test.go
deleted file mode 100644
index 84b37eca3b..0000000000
--- a/event/event_test.go
+++ /dev/null
@@ -1,218 +0,0 @@
-// Copyright 2014 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event
-
-import (
- "math/rand"
- "sync"
- "testing"
- "time"
-)
-
-type testEvent int
-
-func TestSubCloseUnsub(t *testing.T) {
- // the point of this test is **not** to panic
- var mux TypeMux
- mux.Stop()
- sub := mux.Subscribe(0)
- sub.Unsubscribe()
-}
-
-func TestSub(t *testing.T) {
- mux := new(TypeMux)
- defer mux.Stop()
-
- sub := mux.Subscribe(testEvent(0))
- go func() {
- if err := mux.Post(testEvent(5)); err != nil {
- t.Errorf("Post returned unexpected error: %v", err)
- }
- }()
- ev := <-sub.Chan()
-
- if ev.Data.(testEvent) != testEvent(5) {
- t.Errorf("Got %v (%T), expected event %v (%T)",
- ev, ev, testEvent(5), testEvent(5))
- }
-}
-
-func TestMuxErrorAfterStop(t *testing.T) {
- mux := new(TypeMux)
- mux.Stop()
-
- sub := mux.Subscribe(testEvent(0))
- if _, isopen := <-sub.Chan(); isopen {
- t.Errorf("subscription channel was not closed")
- }
- if err := mux.Post(testEvent(0)); err != ErrMuxClosed {
- t.Errorf("Post error mismatch, got: %s, expected: %s", err, ErrMuxClosed)
- }
-}
-
-func TestUnsubscribeUnblockPost(t *testing.T) {
- mux := new(TypeMux)
- defer mux.Stop()
-
- sub := mux.Subscribe(testEvent(0))
- unblocked := make(chan bool)
- go func() {
- mux.Post(testEvent(5))
- unblocked <- true
- }()
-
- select {
- case <-unblocked:
- t.Errorf("Post returned before Unsubscribe")
- default:
- sub.Unsubscribe()
- <-unblocked
- }
-}
-
-func TestSubscribeDuplicateType(t *testing.T) {
- mux := new(TypeMux)
- expected := "event: duplicate type event.testEvent in Subscribe"
-
- defer func() {
- err := recover()
- if err == nil {
- t.Errorf("Subscribe didn't panic for duplicate type")
- } else if err != expected {
- t.Errorf("panic mismatch: got %#v, expected %#v", err, expected)
- }
- }()
- mux.Subscribe(testEvent(1), testEvent(2))
-}
-
-func TestMuxConcurrent(t *testing.T) {
- mux := new(TypeMux)
- defer mux.Stop()
-
- recv := make(chan int)
- poster := func() {
- for {
- err := mux.Post(testEvent(0))
- if err != nil {
- return
- }
- }
- }
- sub := func(i int) {
- time.Sleep(time.Duration(rand.Intn(99)) * time.Millisecond)
- sub := mux.Subscribe(testEvent(0))
- <-sub.Chan()
- sub.Unsubscribe()
- recv <- i
- }
-
- go poster()
- go poster()
- go poster()
- nsubs := 1000
- for i := 0; i < nsubs; i++ {
- go sub(i)
- }
-
- // wait until everyone has been served
- counts := make(map[int]int, nsubs)
- for i := 0; i < nsubs; i++ {
- counts[<-recv]++
- }
- for i, count := range counts {
- if count != 1 {
- t.Errorf("receiver %d called %d times, expected only 1 call", i, count)
- }
- }
-}
-
-func emptySubscriber(mux *TypeMux) {
- s := mux.Subscribe(testEvent(0))
- go func() {
- for range s.Chan() {
- }
- }()
-}
-
-func BenchmarkPost1000(b *testing.B) {
- var (
- mux = new(TypeMux)
- subscribed, done sync.WaitGroup
- nsubs = 1000
- )
- subscribed.Add(nsubs)
- done.Add(nsubs)
- for i := 0; i < nsubs; i++ {
- go func() {
- s := mux.Subscribe(testEvent(0))
- subscribed.Done()
- for range s.Chan() {
- }
- done.Done()
- }()
- }
- subscribed.Wait()
-
- // The actual benchmark.
- b.ResetTimer()
- for i := 0; i < b.N; i++ {
- mux.Post(testEvent(0))
- }
-
- b.StopTimer()
- mux.Stop()
- done.Wait()
-}
-
-func BenchmarkPostConcurrent(b *testing.B) {
- var mux = new(TypeMux)
- defer mux.Stop()
- emptySubscriber(mux)
- emptySubscriber(mux)
- emptySubscriber(mux)
-
- var wg sync.WaitGroup
- poster := func() {
- for i := 0; i < b.N; i++ {
- mux.Post(testEvent(0))
- }
- wg.Done()
- }
- wg.Add(5)
- for i := 0; i < 5; i++ {
- go poster()
- }
- wg.Wait()
-}
-
-// for comparison
-func BenchmarkChanSend(b *testing.B) {
- c := make(chan interface{})
- defer close(c)
- closed := make(chan struct{})
- go func() {
- for range c {
- }
- }()
-
- for i := 0; i < b.N; i++ {
- select {
- case c <- i:
- case <-closed:
- }
- }
-}
diff --git a/event/example_feed_test.go b/event/example_feed_test.go
deleted file mode 100644
index 9b5ad50df5..0000000000
--- a/event/example_feed_test.go
+++ /dev/null
@@ -1,73 +0,0 @@
-// Copyright 2016 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event_test
-
-import (
- "fmt"
-
- "github.com/ethereum/go-ethereum/event"
-)
-
-func ExampleFeed_acknowledgedEvents() {
- // This example shows how the return value of Send can be used for request/reply
- // interaction between event consumers and producers.
- var feed event.Feed
- type ackedEvent struct {
- i int
- ack chan<- struct{}
- }
-
- // Consumers wait for events on the feed and acknowledge processing.
- done := make(chan struct{})
- defer close(done)
- for i := 0; i < 3; i++ {
- ch := make(chan ackedEvent, 100)
- sub := feed.Subscribe(ch)
- go func() {
- defer sub.Unsubscribe()
- for {
- select {
- case ev := <-ch:
- fmt.Println(ev.i) // "process" the event
- ev.ack <- struct{}{}
- case <-done:
- return
- }
- }
- }()
- }
-
- // The producer sends values of type ackedEvent with increasing values of i.
- // It waits for all consumers to acknowledge before sending the next event.
- for i := 0; i < 3; i++ {
- acksignal := make(chan struct{})
- n := feed.Send(ackedEvent{i, acksignal})
- for ack := 0; ack < n; ack++ {
- <-acksignal
- }
- }
- // Output:
- // 0
- // 0
- // 0
- // 1
- // 1
- // 1
- // 2
- // 2
- // 2
-}
diff --git a/event/example_scope_test.go b/event/example_scope_test.go
deleted file mode 100644
index 825a8deeac..0000000000
--- a/event/example_scope_test.go
+++ /dev/null
@@ -1,128 +0,0 @@
-// Copyright 2016 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event_test
-
-import (
- "fmt"
- "sync"
-
- "github.com/ethereum/go-ethereum/event"
-)
-
-// This example demonstrates how SubscriptionScope can be used to control the lifetime of
-// subscriptions.
-//
-// Our example program consists of two servers, each of which performs a calculation when
-// requested. The servers also allow subscribing to results of all computations.
-type divServer struct{ results event.Feed }
-type mulServer struct{ results event.Feed }
-
-func (s *divServer) do(a, b int) int {
- r := a / b
- s.results.Send(r)
- return r
-}
-
-func (s *mulServer) do(a, b int) int {
- r := a * b
- s.results.Send(r)
- return r
-}
-
-// The servers are contained in an App. The app controls the servers and exposes them
-// through its API.
-type App struct {
- divServer
- mulServer
- scope event.SubscriptionScope
-}
-
-func (s *App) Calc(op byte, a, b int) int {
- switch op {
- case '/':
- return s.divServer.do(a, b)
- case '*':
- return s.mulServer.do(a, b)
- default:
- panic("invalid op")
- }
-}
-
-// The app's SubscribeResults method starts sending calculation results to the given
-// channel. Subscriptions created through this method are tied to the lifetime of the App
-// because they are registered in the scope.
-func (s *App) SubscribeResults(op byte, ch chan<- int) event.Subscription {
- switch op {
- case '/':
- return s.scope.Track(s.divServer.results.Subscribe(ch))
- case '*':
- return s.scope.Track(s.mulServer.results.Subscribe(ch))
- default:
- panic("invalid op")
- }
-}
-
-// Stop stops the App, closing all subscriptions created through SubscribeResults.
-func (s *App) Stop() {
- s.scope.Close()
-}
-
-func ExampleSubscriptionScope() {
- // Create the app.
- var (
- app App
- wg sync.WaitGroup
- divs = make(chan int)
- muls = make(chan int)
- )
-
- // Run a subscriber in the background.
- divsub := app.SubscribeResults('/', divs)
- mulsub := app.SubscribeResults('*', muls)
- wg.Add(1)
- go func() {
- defer wg.Done()
- defer fmt.Println("subscriber exited")
- defer divsub.Unsubscribe()
- defer mulsub.Unsubscribe()
- for {
- select {
- case result := <-divs:
- fmt.Println("division happened:", result)
- case result := <-muls:
- fmt.Println("multiplication happened:", result)
- case <-divsub.Err():
- return
- case <-mulsub.Err():
- return
- }
- }
- }()
-
- // Interact with the app.
- app.Calc('/', 22, 11)
- app.Calc('*', 3, 4)
-
- // Stop the app. This shuts down the subscriptions, causing the subscriber to exit.
- app.Stop()
- wg.Wait()
-
- // Output:
- // division happened: 2
- // multiplication happened: 12
- // subscriber exited
-}
diff --git a/event/example_subscription_test.go b/event/example_subscription_test.go
deleted file mode 100644
index 5c76b55d98..0000000000
--- a/event/example_subscription_test.go
+++ /dev/null
@@ -1,56 +0,0 @@
-// Copyright 2016 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event_test
-
-import (
- "fmt"
-
- "github.com/ethereum/go-ethereum/event"
-)
-
-func ExampleNewSubscription() {
- // Create a subscription that sends 10 integers on ch.
- ch := make(chan int)
- sub := event.NewSubscription(func(quit <-chan struct{}) error {
- for i := 0; i < 10; i++ {
- select {
- case ch <- i:
- case <-quit:
- fmt.Println("unsubscribed")
- return nil
- }
- }
- return nil
- })
-
- // This is the consumer. It reads 5 integers, then aborts the subscription.
- // Note that Unsubscribe waits until the producer has shut down.
- for i := range ch {
- fmt.Println(i)
- if i == 4 {
- sub.Unsubscribe()
- break
- }
- }
- // Output:
- // 0
- // 1
- // 2
- // 3
- // 4
- // unsubscribed
-}
diff --git a/event/example_test.go b/event/example_test.go
deleted file mode 100644
index 29938e8539..0000000000
--- a/event/example_test.go
+++ /dev/null
@@ -1,58 +0,0 @@
-// Copyright 2014 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event
-
-import "fmt"
-
-func ExampleTypeMux() {
- type someEvent struct{ I int }
- type otherEvent struct{ S string }
- type yetAnotherEvent struct{ X, Y int }
-
- var mux TypeMux
-
- // Start a subscriber.
- done := make(chan struct{})
- sub := mux.Subscribe(someEvent{}, otherEvent{})
- go func() {
- for event := range sub.Chan() {
- fmt.Printf("Received: %#v\n", event.Data)
- }
- fmt.Println("done")
- close(done)
- }()
-
- // Post some events.
- mux.Post(someEvent{5})
- mux.Post(yetAnotherEvent{X: 3, Y: 4})
- mux.Post(someEvent{6})
- mux.Post(otherEvent{"whoa"})
-
- // Stop closes all subscription channels.
- // The subscriber goroutine will print "done"
- // and exit.
- mux.Stop()
-
- // Wait for subscriber to return.
- <-done
-
- // Output:
- // Received: event.someEvent{I:5}
- // Received: event.someEvent{I:6}
- // Received: event.otherEvent{S:"whoa"}
- // done
-}
diff --git a/event/feed.go b/event/feed.go
deleted file mode 100644
index d94bd820f0..0000000000
--- a/event/feed.go
+++ /dev/null
@@ -1,238 +0,0 @@
-// Copyright 2016 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event
-
-import (
- "errors"
- "reflect"
- "sync"
-)
-
-var errBadChannel = errors.New("event: Subscribe argument does not have sendable channel type")
-
-// Feed implements one-to-many subscriptions where the carrier of events is a channel.
-// Values sent to a Feed are delivered to all subscribed channels simultaneously.
-//
-// Feeds can only be used with a single type. The type is determined by the first Send or
-// Subscribe operation. Subsequent calls to these methods panic if the type does not
-// match.
-//
-// The zero value is ready to use.
-type Feed struct {
- once sync.Once // ensures that init only runs once
- sendLock chan struct{} // sendLock has a one-element buffer and is empty when held.It protects sendCases.
- removeSub chan interface{} // interrupts Send
- sendCases caseList // the active set of select cases used by Send
-
- // The inbox holds newly subscribed channels until they are added to sendCases.
- mu sync.Mutex
- inbox caseList
- etype reflect.Type
-}
-
-// This is the index of the first actual subscription channel in sendCases.
-// sendCases[0] is a SelectRecv case for the removeSub channel.
-const firstSubSendCase = 1
-
-type feedTypeError struct {
- got, want reflect.Type
- op string
-}
-
-func (e feedTypeError) Error() string {
- return "event: wrong type in " + e.op + " got " + e.got.String() + ", want " + e.want.String()
-}
-
-func (f *Feed) init(etype reflect.Type) {
- f.etype = etype
- f.removeSub = make(chan interface{})
- f.sendLock = make(chan struct{}, 1)
- f.sendLock <- struct{}{}
- f.sendCases = caseList{{Chan: reflect.ValueOf(f.removeSub), Dir: reflect.SelectRecv}}
-}
-
-// Subscribe adds a channel to the feed. Future sends will be delivered on the channel
-// until the subscription is canceled. All channels added must have the same element type.
-//
-// The channel should have ample buffer space to avoid blocking other subscribers.
-// Slow subscribers are not dropped.
-func (f *Feed) Subscribe(channel interface{}) Subscription {
- chanval := reflect.ValueOf(channel)
- chantyp := chanval.Type()
- if chantyp.Kind() != reflect.Chan || chantyp.ChanDir()&reflect.SendDir == 0 {
- panic(errBadChannel)
- }
- sub := &feedSub{feed: f, channel: chanval, err: make(chan error, 1)}
-
- f.once.Do(func() { f.init(chantyp.Elem()) })
- if f.etype != chantyp.Elem() {
- panic(feedTypeError{op: "Subscribe", got: chantyp, want: reflect.ChanOf(reflect.SendDir, f.etype)})
- }
-
- f.mu.Lock()
- defer f.mu.Unlock()
- // Add the select case to the inbox.
- // The next Send will add it to f.sendCases.
- cas := reflect.SelectCase{Dir: reflect.SelectSend, Chan: chanval}
- f.inbox = append(f.inbox, cas)
- return sub
-}
-
-func (f *Feed) remove(sub *feedSub) {
- // Delete from inbox first, which covers channels
- // that have not been added to f.sendCases yet.
- ch := sub.channel.Interface()
- f.mu.Lock()
- index := f.inbox.find(ch)
- if index != -1 {
- f.inbox = f.inbox.delete(index)
- f.mu.Unlock()
- return
- }
- f.mu.Unlock()
-
- select {
- case f.removeSub <- ch:
- // Send will remove the channel from f.sendCases.
- case <-f.sendLock:
- // No Send is in progress, delete the channel now that we have the send lock.
- f.sendCases = f.sendCases.delete(f.sendCases.find(ch))
- f.sendLock <- struct{}{}
- }
-}
-
-// Send delivers to all subscribed channels simultaneously.
-// It returns the number of subscribers that the value was sent to.
-func (f *Feed) Send(value interface{}) (nsent int) {
- rvalue := reflect.ValueOf(value)
-
- f.once.Do(func() { f.init(rvalue.Type()) })
- if f.etype != rvalue.Type() {
- panic(feedTypeError{op: "Send", got: rvalue.Type(), want: f.etype})
- }
-
- <-f.sendLock
-
- // Add new cases from the inbox after taking the send lock.
- f.mu.Lock()
- f.sendCases = append(f.sendCases, f.inbox...)
- f.inbox = nil
- f.mu.Unlock()
-
- // Set the sent value on all channels.
- for i := firstSubSendCase; i < len(f.sendCases); i++ {
- f.sendCases[i].Send = rvalue
- }
-
- // Send until all channels except removeSub have been chosen. 'cases' tracks a prefix
- // of sendCases. When a send succeeds, the corresponding case moves to the end of
- // 'cases' and it shrinks by one element.
- cases := f.sendCases
- for {
- // Fast path: try sending without blocking before adding to the select set.
- // This should usually succeed if subscribers are fast enough and have free
- // buffer space.
- for i := firstSubSendCase; i < len(cases); i++ {
- if cases[i].Chan.TrySend(rvalue) {
- nsent++
- cases = cases.deactivate(i)
- i--
- }
- }
- if len(cases) == firstSubSendCase {
- break
- }
- // Select on all the receivers, waiting for them to unblock.
- chosen, recv, _ := reflect.Select(cases)
- if chosen == 0 /* <-f.removeSub */ {
- index := f.sendCases.find(recv.Interface())
- f.sendCases = f.sendCases.delete(index)
- if index >= 0 && index < len(cases) {
- // Shrink 'cases' too because the removed case was still active.
- cases = f.sendCases[:len(cases)-1]
- }
- } else {
- cases = cases.deactivate(chosen)
- nsent++
- }
- }
-
- // Forget about the sent value and hand off the send lock.
- for i := firstSubSendCase; i < len(f.sendCases); i++ {
- f.sendCases[i].Send = reflect.Value{}
- }
- f.sendLock <- struct{}{}
- return nsent
-}
-
-type feedSub struct {
- feed *Feed
- channel reflect.Value
- errOnce sync.Once
- err chan error
-}
-
-func (sub *feedSub) Unsubscribe() {
- sub.errOnce.Do(func() {
- sub.feed.remove(sub)
- close(sub.err)
- })
-}
-
-func (sub *feedSub) Err() <-chan error {
- return sub.err
-}
-
-type caseList []reflect.SelectCase
-
-// find returns the index of a case containing the given channel.
-func (cs caseList) find(channel interface{}) int {
- for i, cas := range cs {
- if cas.Chan.Interface() == channel {
- return i
- }
- }
- return -1
-}
-
-// delete removes the given case from cs.
-func (cs caseList) delete(index int) caseList {
- return append(cs[:index], cs[index+1:]...)
-}
-
-// deactivate moves the case at index into the non-accessible portion of the cs slice.
-func (cs caseList) deactivate(index int) caseList {
- last := len(cs) - 1
- cs[index], cs[last] = cs[last], cs[index]
- return cs[:last]
-}
-
-// func (cs caseList) String() string {
-// s := "["
-// for i, cas := range cs {
-// if i != 0 {
-// s += ", "
-// }
-// switch cas.Dir {
-// case reflect.SelectSend:
-// s += fmt.Sprintf("%v<-", cas.Chan.Interface())
-// case reflect.SelectRecv:
-// s += fmt.Sprintf("<-%v", cas.Chan.Interface())
-// }
-// }
-// return s + "]"
-// }
diff --git a/event/feed_test.go b/event/feed_test.go
deleted file mode 100644
index 74e8587a87..0000000000
--- a/event/feed_test.go
+++ /dev/null
@@ -1,335 +0,0 @@
-// Copyright 2016 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event
-
-import (
- "errors"
- "fmt"
- "reflect"
- "sync"
- "testing"
- "time"
-)
-
-func TestFeedPanics(t *testing.T) {
- {
- var f Feed
- f.Send(2)
- want := feedTypeError{op: "Send", got: reflect.TypeOf(uint64(0)), want: reflect.TypeOf(0)}
- if err := checkPanic(want, func() { f.Send(uint64(2)) }); err != nil {
- t.Error(err)
- }
- }
- {
- var f Feed
- ch := make(chan int)
- f.Subscribe(ch)
- want := feedTypeError{op: "Send", got: reflect.TypeOf(uint64(0)), want: reflect.TypeOf(0)}
- if err := checkPanic(want, func() { f.Send(uint64(2)) }); err != nil {
- t.Error(err)
- }
- }
- {
- var f Feed
- f.Send(2)
- want := feedTypeError{op: "Subscribe", got: reflect.TypeOf(make(chan uint64)), want: reflect.TypeOf(make(chan<- int))}
- if err := checkPanic(want, func() { f.Subscribe(make(chan uint64)) }); err != nil {
- t.Error(err)
- }
- }
- {
- var f Feed
- if err := checkPanic(errBadChannel, func() { f.Subscribe(make(<-chan int)) }); err != nil {
- t.Error(err)
- }
- }
- {
- var f Feed
- if err := checkPanic(errBadChannel, func() { f.Subscribe(0) }); err != nil {
- t.Error(err)
- }
- }
-}
-
-func checkPanic(want error, fn func()) (err error) {
- defer func() {
- panic := recover()
- if panic == nil {
- err = errors.New("didn't panic")
- } else if !reflect.DeepEqual(panic, want) {
- err = fmt.Errorf("panicked with wrong error: got %q, want %q", panic, want)
- }
- }()
- fn()
- return nil
-}
-
-func TestFeed(t *testing.T) {
- var feed Feed
- var done, subscribed sync.WaitGroup
- subscriber := func(i int) {
- defer done.Done()
-
- subchan := make(chan int)
- sub := feed.Subscribe(subchan)
- timeout := time.NewTimer(2 * time.Second)
- defer timeout.Stop()
- subscribed.Done()
-
- select {
- case v := <-subchan:
- if v != 1 {
- t.Errorf("%d: received value %d, want 1", i, v)
- }
- case <-timeout.C:
- t.Errorf("%d: receive timeout", i)
- }
-
- sub.Unsubscribe()
- select {
- case _, ok := <-sub.Err():
- if ok {
- t.Errorf("%d: error channel not closed after unsubscribe", i)
- }
- case <-timeout.C:
- t.Errorf("%d: unsubscribe timeout", i)
- }
- }
-
- const n = 1000
- done.Add(n)
- subscribed.Add(n)
- for i := 0; i < n; i++ {
- go subscriber(i)
- }
- subscribed.Wait()
- if nsent := feed.Send(1); nsent != n {
- t.Errorf("first send delivered %d times, want %d", nsent, n)
- }
- if nsent := feed.Send(2); nsent != 0 {
- t.Errorf("second send delivered %d times, want 0", nsent)
- }
- done.Wait()
-}
-
-func TestFeedSubscribeSameChannel(t *testing.T) {
- var (
- feed Feed
- done sync.WaitGroup
- ch = make(chan int)
- sub1 = feed.Subscribe(ch)
- sub2 = feed.Subscribe(ch)
- _ = feed.Subscribe(ch)
- )
- expectSends := func(value, n int) {
- if nsent := feed.Send(value); nsent != n {
- t.Errorf("send delivered %d times, want %d", nsent, n)
- }
- done.Done()
- }
- expectRecv := func(wantValue, n int) {
- for i := 0; i < n; i++ {
- if v := <-ch; v != wantValue {
- t.Errorf("received %d, want %d", v, wantValue)
- }
- }
- }
-
- done.Add(1)
- go expectSends(1, 3)
- expectRecv(1, 3)
- done.Wait()
-
- sub1.Unsubscribe()
-
- done.Add(1)
- go expectSends(2, 2)
- expectRecv(2, 2)
- done.Wait()
-
- sub2.Unsubscribe()
-
- done.Add(1)
- go expectSends(3, 1)
- expectRecv(3, 1)
- done.Wait()
-}
-
-func TestFeedSubscribeBlockedPost(t *testing.T) {
- var (
- feed Feed
- nsends = 2000
- ch1 = make(chan int)
- ch2 = make(chan int)
- wg sync.WaitGroup
- )
- defer wg.Wait()
-
- feed.Subscribe(ch1)
- wg.Add(nsends)
- for i := 0; i < nsends; i++ {
- go func() {
- feed.Send(99)
- wg.Done()
- }()
- }
-
- sub2 := feed.Subscribe(ch2)
- defer sub2.Unsubscribe()
-
- // We're done when ch1 has received N times.
- // The number of receives on ch2 depends on scheduling.
- for i := 0; i < nsends; {
- select {
- case <-ch1:
- i++
- case <-ch2:
- }
- }
-}
-
-func TestFeedUnsubscribeBlockedPost(t *testing.T) {
- var (
- feed Feed
- nsends = 200
- chans = make([]chan int, 2000)
- subs = make([]Subscription, len(chans))
- bchan = make(chan int)
- bsub = feed.Subscribe(bchan)
- wg sync.WaitGroup
- )
- for i := range chans {
- chans[i] = make(chan int, nsends)
- }
-
- // Queue up some Sends. None of these can make progress while bchan isn't read.
- wg.Add(nsends)
- for i := 0; i < nsends; i++ {
- go func() {
- feed.Send(99)
- wg.Done()
- }()
- }
- // Subscribe the other channels.
- for i, ch := range chans {
- subs[i] = feed.Subscribe(ch)
- }
- // Unsubscribe them again.
- for _, sub := range subs {
- sub.Unsubscribe()
- }
- // Unblock the Sends.
- bsub.Unsubscribe()
- wg.Wait()
-}
-
-// Checks that unsubscribing a channel during Send works even if that
-// channel has already been sent on.
-func TestFeedUnsubscribeSentChan(t *testing.T) {
- var (
- feed Feed
- ch1 = make(chan int)
- ch2 = make(chan int)
- sub1 = feed.Subscribe(ch1)
- sub2 = feed.Subscribe(ch2)
- wg sync.WaitGroup
- )
- defer sub2.Unsubscribe()
-
- wg.Add(1)
- go func() {
- feed.Send(0)
- wg.Done()
- }()
-
- // Wait for the value on ch1.
- <-ch1
- // Unsubscribe ch1, removing it from the send cases.
- sub1.Unsubscribe()
-
- // Receive ch2, finishing Send.
- <-ch2
- wg.Wait()
-
- // Send again. This should send to ch2 only, so the wait group will unblock
- // as soon as a value is received on ch2.
- wg.Add(1)
- go func() {
- feed.Send(0)
- wg.Done()
- }()
- <-ch2
- wg.Wait()
-}
-
-func TestFeedUnsubscribeFromInbox(t *testing.T) {
- var (
- feed Feed
- ch1 = make(chan int)
- ch2 = make(chan int)
- sub1 = feed.Subscribe(ch1)
- sub2 = feed.Subscribe(ch1)
- sub3 = feed.Subscribe(ch2)
- )
- if len(feed.inbox) != 3 {
- t.Errorf("inbox length != 3 after subscribe")
- }
- if len(feed.sendCases) != 1 {
- t.Errorf("sendCases is non-empty after unsubscribe")
- }
-
- sub1.Unsubscribe()
- sub2.Unsubscribe()
- sub3.Unsubscribe()
- if len(feed.inbox) != 0 {
- t.Errorf("inbox is non-empty after unsubscribe")
- }
- if len(feed.sendCases) != 1 {
- t.Errorf("sendCases is non-empty after unsubscribe")
- }
-}
-
-func BenchmarkFeedSend1000(b *testing.B) {
- var (
- done sync.WaitGroup
- feed Feed
- nsubs = 1000
- )
- subscriber := func(ch <-chan int) {
- for i := 0; i < b.N; i++ {
- <-ch
- }
- done.Done()
- }
- done.Add(nsubs)
- for i := 0; i < nsubs; i++ {
- ch := make(chan int, 200)
- feed.Subscribe(ch)
- go subscriber(ch)
- }
-
- // The actual benchmark.
- b.ResetTimer()
- for i := 0; i < b.N; i++ {
- if feed.Send(i) != nsubs {
- panic("wrong number of sends")
- }
- }
-
- b.StopTimer()
- done.Wait()
-}
diff --git a/event/feedof.go b/event/feedof.go
deleted file mode 100644
index 4a24e37f12..0000000000
--- a/event/feedof.go
+++ /dev/null
@@ -1,164 +0,0 @@
-// Copyright 2022 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event
-
-import (
- "reflect"
- "sync"
-)
-
-// FeedOf implements one-to-many subscriptions where the carrier of events is a channel.
-// Values sent to a Feed are delivered to all subscribed channels simultaneously.
-//
-// The zero value is ready to use.
-type FeedOf[T any] struct {
- once sync.Once // ensures that init only runs once
- sendLock chan struct{} // sendLock has a one-element buffer and is empty when held.It protects sendCases.
- removeSub chan chan<- T // interrupts Send
- sendCases caseList // the active set of select cases used by Send
-
- // The inbox holds newly subscribed channels until they are added to sendCases.
- mu sync.Mutex
- inbox caseList
-}
-
-func (f *FeedOf[T]) init() {
- f.removeSub = make(chan chan<- T)
- f.sendLock = make(chan struct{}, 1)
- f.sendLock <- struct{}{}
- f.sendCases = caseList{{Chan: reflect.ValueOf(f.removeSub), Dir: reflect.SelectRecv}}
-}
-
-// Subscribe adds a channel to the feed. Future sends will be delivered on the channel
-// until the subscription is canceled.
-//
-// The channel should have ample buffer space to avoid blocking other subscribers. Slow
-// subscribers are not dropped.
-func (f *FeedOf[T]) Subscribe(channel chan<- T) Subscription {
- f.once.Do(f.init)
-
- chanval := reflect.ValueOf(channel)
- sub := &feedOfSub[T]{feed: f, channel: channel, err: make(chan error, 1)}
-
- // Add the select case to the inbox.
- // The next Send will add it to f.sendCases.
- f.mu.Lock()
- defer f.mu.Unlock()
- cas := reflect.SelectCase{Dir: reflect.SelectSend, Chan: chanval}
- f.inbox = append(f.inbox, cas)
- return sub
-}
-
-func (f *FeedOf[T]) remove(sub *feedOfSub[T]) {
- // Delete from inbox first, which covers channels
- // that have not been added to f.sendCases yet.
- f.mu.Lock()
- index := f.inbox.find(sub.channel)
- if index != -1 {
- f.inbox = f.inbox.delete(index)
- f.mu.Unlock()
- return
- }
- f.mu.Unlock()
-
- select {
- case f.removeSub <- sub.channel:
- // Send will remove the channel from f.sendCases.
- case <-f.sendLock:
- // No Send is in progress, delete the channel now that we have the send lock.
- f.sendCases = f.sendCases.delete(f.sendCases.find(sub.channel))
- f.sendLock <- struct{}{}
- }
-}
-
-// Send delivers to all subscribed channels simultaneously.
-// It returns the number of subscribers that the value was sent to.
-func (f *FeedOf[T]) Send(value T) (nsent int) {
- rvalue := reflect.ValueOf(value)
-
- f.once.Do(f.init)
- <-f.sendLock
-
- // Add new cases from the inbox after taking the send lock.
- f.mu.Lock()
- f.sendCases = append(f.sendCases, f.inbox...)
- f.inbox = nil
- f.mu.Unlock()
-
- // Set the sent value on all channels.
- for i := firstSubSendCase; i < len(f.sendCases); i++ {
- f.sendCases[i].Send = rvalue
- }
-
- // Send until all channels except removeSub have been chosen. 'cases' tracks a prefix
- // of sendCases. When a send succeeds, the corresponding case moves to the end of
- // 'cases' and it shrinks by one element.
- cases := f.sendCases
- for {
- // Fast path: try sending without blocking before adding to the select set.
- // This should usually succeed if subscribers are fast enough and have free
- // buffer space.
- for i := firstSubSendCase; i < len(cases); i++ {
- if cases[i].Chan.TrySend(rvalue) {
- nsent++
- cases = cases.deactivate(i)
- i--
- }
- }
- if len(cases) == firstSubSendCase {
- break
- }
- // Select on all the receivers, waiting for them to unblock.
- chosen, recv, _ := reflect.Select(cases)
- if chosen == 0 /* <-f.removeSub */ {
- index := f.sendCases.find(recv.Interface())
- f.sendCases = f.sendCases.delete(index)
- if index >= 0 && index < len(cases) {
- // Shrink 'cases' too because the removed case was still active.
- cases = f.sendCases[:len(cases)-1]
- }
- } else {
- cases = cases.deactivate(chosen)
- nsent++
- }
- }
-
- // Forget about the sent value and hand off the send lock.
- for i := firstSubSendCase; i < len(f.sendCases); i++ {
- f.sendCases[i].Send = reflect.Value{}
- }
- f.sendLock <- struct{}{}
- return nsent
-}
-
-type feedOfSub[T any] struct {
- feed *FeedOf[T]
- channel chan<- T
- errOnce sync.Once
- err chan error
-}
-
-func (sub *feedOfSub[T]) Unsubscribe() {
- sub.errOnce.Do(func() {
- sub.feed.remove(sub)
- close(sub.err)
- })
-}
-
-func (sub *feedOfSub[T]) Err() <-chan error {
- return sub.err
-}
diff --git a/event/feedof_test.go b/event/feedof_test.go
deleted file mode 100644
index 846afc9ee1..0000000000
--- a/event/feedof_test.go
+++ /dev/null
@@ -1,279 +0,0 @@
-// Copyright 2016 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event
-
-import (
- "sync"
- "testing"
- "time"
-)
-
-func TestFeedOf(t *testing.T) {
- var feed FeedOf[int]
- var done, subscribed sync.WaitGroup
- subscriber := func(i int) {
- defer done.Done()
-
- subchan := make(chan int)
- sub := feed.Subscribe(subchan)
- timeout := time.NewTimer(2 * time.Second)
- defer timeout.Stop()
- subscribed.Done()
-
- select {
- case v := <-subchan:
- if v != 1 {
- t.Errorf("%d: received value %d, want 1", i, v)
- }
- case <-timeout.C:
- t.Errorf("%d: receive timeout", i)
- }
-
- sub.Unsubscribe()
- select {
- case _, ok := <-sub.Err():
- if ok {
- t.Errorf("%d: error channel not closed after unsubscribe", i)
- }
- case <-timeout.C:
- t.Errorf("%d: unsubscribe timeout", i)
- }
- }
-
- const n = 1000
- done.Add(n)
- subscribed.Add(n)
- for i := 0; i < n; i++ {
- go subscriber(i)
- }
- subscribed.Wait()
- if nsent := feed.Send(1); nsent != n {
- t.Errorf("first send delivered %d times, want %d", nsent, n)
- }
- if nsent := feed.Send(2); nsent != 0 {
- t.Errorf("second send delivered %d times, want 0", nsent)
- }
- done.Wait()
-}
-
-func TestFeedOfSubscribeSameChannel(t *testing.T) {
- var (
- feed FeedOf[int]
- done sync.WaitGroup
- ch = make(chan int)
- sub1 = feed.Subscribe(ch)
- sub2 = feed.Subscribe(ch)
- _ = feed.Subscribe(ch)
- )
- expectSends := func(value, n int) {
- if nsent := feed.Send(value); nsent != n {
- t.Errorf("send delivered %d times, want %d", nsent, n)
- }
- done.Done()
- }
- expectRecv := func(wantValue, n int) {
- for i := 0; i < n; i++ {
- if v := <-ch; v != wantValue {
- t.Errorf("received %d, want %d", v, wantValue)
- }
- }
- }
-
- done.Add(1)
- go expectSends(1, 3)
- expectRecv(1, 3)
- done.Wait()
-
- sub1.Unsubscribe()
-
- done.Add(1)
- go expectSends(2, 2)
- expectRecv(2, 2)
- done.Wait()
-
- sub2.Unsubscribe()
-
- done.Add(1)
- go expectSends(3, 1)
- expectRecv(3, 1)
- done.Wait()
-}
-
-func TestFeedOfSubscribeBlockedPost(t *testing.T) {
- var (
- feed FeedOf[int]
- nsends = 2000
- ch1 = make(chan int)
- ch2 = make(chan int)
- wg sync.WaitGroup
- )
- defer wg.Wait()
-
- feed.Subscribe(ch1)
- wg.Add(nsends)
- for i := 0; i < nsends; i++ {
- go func() {
- feed.Send(99)
- wg.Done()
- }()
- }
-
- sub2 := feed.Subscribe(ch2)
- defer sub2.Unsubscribe()
-
- // We're done when ch1 has received N times.
- // The number of receives on ch2 depends on scheduling.
- for i := 0; i < nsends; {
- select {
- case <-ch1:
- i++
- case <-ch2:
- }
- }
-}
-
-func TestFeedOfUnsubscribeBlockedPost(t *testing.T) {
- var (
- feed FeedOf[int]
- nsends = 200
- chans = make([]chan int, 2000)
- subs = make([]Subscription, len(chans))
- bchan = make(chan int)
- bsub = feed.Subscribe(bchan)
- wg sync.WaitGroup
- )
- for i := range chans {
- chans[i] = make(chan int, nsends)
- }
-
- // Queue up some Sends. None of these can make progress while bchan isn't read.
- wg.Add(nsends)
- for i := 0; i < nsends; i++ {
- go func() {
- feed.Send(99)
- wg.Done()
- }()
- }
- // Subscribe the other channels.
- for i, ch := range chans {
- subs[i] = feed.Subscribe(ch)
- }
- // Unsubscribe them again.
- for _, sub := range subs {
- sub.Unsubscribe()
- }
- // Unblock the Sends.
- bsub.Unsubscribe()
- wg.Wait()
-}
-
-// Checks that unsubscribing a channel during Send works even if that
-// channel has already been sent on.
-func TestFeedOfUnsubscribeSentChan(t *testing.T) {
- var (
- feed FeedOf[int]
- ch1 = make(chan int)
- ch2 = make(chan int)
- sub1 = feed.Subscribe(ch1)
- sub2 = feed.Subscribe(ch2)
- wg sync.WaitGroup
- )
- defer sub2.Unsubscribe()
-
- wg.Add(1)
- go func() {
- feed.Send(0)
- wg.Done()
- }()
-
- // Wait for the value on ch1.
- <-ch1
- // Unsubscribe ch1, removing it from the send cases.
- sub1.Unsubscribe()
-
- // Receive ch2, finishing Send.
- <-ch2
- wg.Wait()
-
- // Send again. This should send to ch2 only, so the wait group will unblock
- // as soon as a value is received on ch2.
- wg.Add(1)
- go func() {
- feed.Send(0)
- wg.Done()
- }()
- <-ch2
- wg.Wait()
-}
-
-func TestFeedOfUnsubscribeFromInbox(t *testing.T) {
- var (
- feed FeedOf[int]
- ch1 = make(chan int)
- ch2 = make(chan int)
- sub1 = feed.Subscribe(ch1)
- sub2 = feed.Subscribe(ch1)
- sub3 = feed.Subscribe(ch2)
- )
- if len(feed.inbox) != 3 {
- t.Errorf("inbox length != 3 after subscribe")
- }
- if len(feed.sendCases) != 1 {
- t.Errorf("sendCases is non-empty after unsubscribe")
- }
-
- sub1.Unsubscribe()
- sub2.Unsubscribe()
- sub3.Unsubscribe()
- if len(feed.inbox) != 0 {
- t.Errorf("inbox is non-empty after unsubscribe")
- }
- if len(feed.sendCases) != 1 {
- t.Errorf("sendCases is non-empty after unsubscribe")
- }
-}
-
-func BenchmarkFeedOfSend1000(b *testing.B) {
- var (
- done sync.WaitGroup
- feed FeedOf[int]
- nsubs = 1000
- )
- subscriber := func(ch <-chan int) {
- for i := 0; i < b.N; i++ {
- <-ch
- }
- done.Done()
- }
- done.Add(nsubs)
- for i := 0; i < nsubs; i++ {
- ch := make(chan int, 200)
- feed.Subscribe(ch)
- go subscriber(ch)
- }
-
- // The actual benchmark.
- b.ResetTimer()
- for i := 0; i < b.N; i++ {
- if feed.Send(i) != nsubs {
- panic("wrong number of sends")
- }
- }
-
- b.StopTimer()
- done.Wait()
-}
diff --git a/event/multisub.go b/event/multisub.go
deleted file mode 100644
index 5c8d2df48c..0000000000
--- a/event/multisub.go
+++ /dev/null
@@ -1,50 +0,0 @@
-// Copyright 2023 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event
-
-// JoinSubscriptions joins multiple subscriptions to be able to track them as
-// one entity and collectively cancel them of consume any errors from them.
-func JoinSubscriptions(subs ...Subscription) Subscription {
- return NewSubscription(func(unsubbed <-chan struct{}) error {
- // Unsubscribe all subscriptions before returning
- defer func() {
- for _, sub := range subs {
- sub.Unsubscribe()
- }
- }()
- // Wait for an error on any of the subscriptions and propagate up
- errc := make(chan error, len(subs))
- for i := range subs {
- go func(sub Subscription) {
- select {
- case err := <-sub.Err():
- if err != nil {
- errc <- err
- }
- case <-unsubbed:
- }
- }(subs[i])
- }
-
- select {
- case err := <-errc:
- return err
- case <-unsubbed:
- return nil
- }
- })
-}
diff --git a/event/multisub_test.go b/event/multisub_test.go
deleted file mode 100644
index c92bcfae9b..0000000000
--- a/event/multisub_test.go
+++ /dev/null
@@ -1,175 +0,0 @@
-// Copyright 2023 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event
-
-import (
- "testing"
- "time"
-)
-
-func TestMultisub(t *testing.T) {
- // Create a double subscription and ensure events propagate through
- var (
- feed1 Feed
- feed2 Feed
- )
- sink1 := make(chan int, 1)
- sink2 := make(chan int, 1)
-
- sub1 := feed1.Subscribe(sink1)
- sub2 := feed2.Subscribe(sink2)
-
- sub := JoinSubscriptions(sub1, sub2)
-
- feed1.Send(1)
- select {
- case n := <-sink1:
- if n != 1 {
- t.Errorf("sink 1 delivery mismatch: have %d, want %d", n, 1)
- }
- default:
- t.Error("sink 1 missing delivery")
- }
-
- feed2.Send(2)
- select {
- case n := <-sink2:
- if n != 2 {
- t.Errorf("sink 2 delivery mismatch: have %d, want %d", n, 2)
- }
- default:
- t.Error("sink 2 missing delivery")
- }
- // Unsubscribe and ensure no more events are delivered
- sub.Unsubscribe()
- select {
- case <-sub.Err():
- case <-time.After(50 * time.Millisecond):
- t.Error("multisub didn't propagate closure")
- }
-
- feed1.Send(11)
- select {
- case n := <-sink1:
- t.Errorf("sink 1 unexpected delivery: %d", n)
- default:
- }
-
- feed2.Send(22)
- select {
- case n := <-sink2:
- t.Errorf("sink 2 unexpected delivery: %d", n)
- default:
- }
-}
-
-func TestMutisubPartialUnsubscribe(t *testing.T) {
- // Create a double subscription but terminate one half, ensuring no error
- // is propagated yet up to the outer subscription
- var (
- feed1 Feed
- feed2 Feed
- )
- sink1 := make(chan int, 1)
- sink2 := make(chan int, 1)
-
- sub1 := feed1.Subscribe(sink1)
- sub2 := feed2.Subscribe(sink2)
-
- sub := JoinSubscriptions(sub1, sub2)
-
- sub1.Unsubscribe()
- select {
- case <-sub.Err():
- t.Error("multisub propagated closure")
- case <-time.After(50 * time.Millisecond):
- }
- // Ensure that events cross only the second feed
- feed1.Send(1)
- select {
- case n := <-sink1:
- t.Errorf("sink 1 unexpected delivery: %d", n)
- default:
- }
-
- feed2.Send(2)
- select {
- case n := <-sink2:
- if n != 2 {
- t.Errorf("sink 2 delivery mismatch: have %d, want %d", n, 2)
- }
- default:
- t.Error("sink 2 missing delivery")
- }
- // Unsubscribe and ensure no more events are delivered
- sub.Unsubscribe()
- select {
- case <-sub.Err():
- case <-time.After(50 * time.Millisecond):
- t.Error("multisub didn't propagate closure")
- }
-
- feed1.Send(11)
- select {
- case n := <-sink1:
- t.Errorf("sink 1 unexpected delivery: %d", n)
- default:
- }
-
- feed2.Send(22)
- select {
- case n := <-sink2:
- t.Errorf("sink 2 unexpected delivery: %d", n)
- default:
- }
-}
-
-func TestMultisubFullUnsubscribe(t *testing.T) {
- // Create a double subscription and terminate the multi sub, ensuring an
- // error is propagated up.
- var (
- feed1 Feed
- feed2 Feed
- )
- sink1 := make(chan int, 1)
- sink2 := make(chan int, 1)
-
- sub1 := feed1.Subscribe(sink1)
- sub2 := feed2.Subscribe(sink2)
-
- sub := JoinSubscriptions(sub1, sub2)
- sub.Unsubscribe()
- select {
- case <-sub.Err():
- case <-time.After(50 * time.Millisecond):
- t.Error("multisub didn't propagate closure")
- }
- // Ensure no more events are delivered
- feed1.Send(1)
- select {
- case n := <-sink1:
- t.Errorf("sink 1 unexpected delivery: %d", n)
- default:
- }
-
- feed2.Send(2)
- select {
- case n := <-sink2:
- t.Errorf("sink 2 unexpected delivery: %d", n)
- default:
- }
-}
diff --git a/event/subscription.go b/event/subscription.go
deleted file mode 100644
index 07e059c6db..0000000000
--- a/event/subscription.go
+++ /dev/null
@@ -1,298 +0,0 @@
-// Copyright 2016 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event
-
-import (
- "context"
- "sync"
- "time"
-
- "github.com/ethereum/go-ethereum/common/mclock"
-)
-
-// Subscription represents a stream of events. The carrier of the events is typically a
-// channel, but isn't part of the interface.
-//
-// Subscriptions can fail while established. Failures are reported through an error
-// channel. It receives a value if there is an issue with the subscription (e.g. the
-// network connection delivering the events has been closed). Only one value will ever be
-// sent.
-//
-// The error channel is closed when the subscription ends successfully (i.e. when the
-// source of events is closed). It is also closed when Unsubscribe is called.
-//
-// The Unsubscribe method cancels the sending of events. You must call Unsubscribe in all
-// cases to ensure that resources related to the subscription are released. It can be
-// called any number of times.
-type Subscription interface {
- Err() <-chan error // returns the error channel
- Unsubscribe() // cancels sending of events, closing the error channel
-}
-
-// NewSubscription runs a producer function as a subscription in a new goroutine. The
-// channel given to the producer is closed when Unsubscribe is called. If fn returns an
-// error, it is sent on the subscription's error channel.
-func NewSubscription(producer func(<-chan struct{}) error) Subscription {
- s := &funcSub{unsub: make(chan struct{}), err: make(chan error, 1)}
- go func() {
- defer close(s.err)
- err := producer(s.unsub)
- s.mu.Lock()
- defer s.mu.Unlock()
- if !s.unsubscribed {
- if err != nil {
- s.err <- err
- }
- s.unsubscribed = true
- }
- }()
- return s
-}
-
-type funcSub struct {
- unsub chan struct{}
- err chan error
- mu sync.Mutex
- unsubscribed bool
-}
-
-func (s *funcSub) Unsubscribe() {
- s.mu.Lock()
- if s.unsubscribed {
- s.mu.Unlock()
- return
- }
- s.unsubscribed = true
- close(s.unsub)
- s.mu.Unlock()
- // Wait for producer shutdown.
- <-s.err
-}
-
-func (s *funcSub) Err() <-chan error {
- return s.err
-}
-
-// Resubscribe calls fn repeatedly to keep a subscription established. When the
-// subscription is established, Resubscribe waits for it to fail and calls fn again. This
-// process repeats until Unsubscribe is called or the active subscription ends
-// successfully.
-//
-// Resubscribe applies backoff between calls to fn. The time between calls is adapted
-// based on the error rate, but will never exceed backoffMax.
-func Resubscribe(backoffMax time.Duration, fn ResubscribeFunc) Subscription {
- return ResubscribeErr(backoffMax, func(ctx context.Context, _ error) (Subscription, error) {
- return fn(ctx)
- })
-}
-
-// A ResubscribeFunc attempts to establish a subscription.
-type ResubscribeFunc func(context.Context) (Subscription, error)
-
-// ResubscribeErr calls fn repeatedly to keep a subscription established. When the
-// subscription is established, ResubscribeErr waits for it to fail and calls fn again. This
-// process repeats until Unsubscribe is called or the active subscription ends
-// successfully.
-//
-// The difference between Resubscribe and ResubscribeErr is that with ResubscribeErr,
-// the error of the failing subscription is available to the callback for logging
-// purposes.
-//
-// ResubscribeErr applies backoff between calls to fn. The time between calls is adapted
-// based on the error rate, but will never exceed backoffMax.
-func ResubscribeErr(backoffMax time.Duration, fn ResubscribeErrFunc) Subscription {
- s := &resubscribeSub{
- waitTime: backoffMax / 10,
- backoffMax: backoffMax,
- fn: fn,
- err: make(chan error),
- unsub: make(chan struct{}, 1),
- }
- go s.loop()
- return s
-}
-
-// A ResubscribeErrFunc attempts to establish a subscription.
-// For every call but the first, the second argument to this function is
-// the error that occurred with the previous subscription.
-type ResubscribeErrFunc func(context.Context, error) (Subscription, error)
-
-type resubscribeSub struct {
- fn ResubscribeErrFunc
- err chan error
- unsub chan struct{}
- unsubOnce sync.Once
- lastTry mclock.AbsTime
- lastSubErr error
- waitTime, backoffMax time.Duration
-}
-
-func (s *resubscribeSub) Unsubscribe() {
- s.unsubOnce.Do(func() {
- s.unsub <- struct{}{}
- <-s.err
- })
-}
-
-func (s *resubscribeSub) Err() <-chan error {
- return s.err
-}
-
-func (s *resubscribeSub) loop() {
- defer close(s.err)
- var done bool
- for !done {
- sub := s.subscribe()
- if sub == nil {
- break
- }
- done = s.waitForError(sub)
- sub.Unsubscribe()
- }
-}
-
-func (s *resubscribeSub) subscribe() Subscription {
- subscribed := make(chan error)
- var sub Subscription
- for {
- s.lastTry = mclock.Now()
- ctx, cancel := context.WithCancel(context.Background())
- go func() {
- rsub, err := s.fn(ctx, s.lastSubErr)
- sub = rsub
- subscribed <- err
- }()
- select {
- case err := <-subscribed:
- cancel()
- if err == nil {
- if sub == nil {
- panic("event: ResubscribeFunc returned nil subscription and no error")
- }
- return sub
- }
- // Subscribing failed, wait before launching the next try.
- if s.backoffWait() {
- return nil // unsubscribed during wait
- }
- case <-s.unsub:
- cancel()
- <-subscribed // avoid leaking the s.fn goroutine.
- return nil
- }
- }
-}
-
-func (s *resubscribeSub) waitForError(sub Subscription) bool {
- defer sub.Unsubscribe()
- select {
- case err := <-sub.Err():
- s.lastSubErr = err
- return err == nil
- case <-s.unsub:
- return true
- }
-}
-
-func (s *resubscribeSub) backoffWait() bool {
- if time.Duration(mclock.Now()-s.lastTry) > s.backoffMax {
- s.waitTime = s.backoffMax / 10
- } else {
- s.waitTime *= 2
- if s.waitTime > s.backoffMax {
- s.waitTime = s.backoffMax
- }
- }
-
- t := time.NewTimer(s.waitTime)
- defer t.Stop()
- select {
- case <-t.C:
- return false
- case <-s.unsub:
- return true
- }
-}
-
-// SubscriptionScope provides a facility to unsubscribe multiple subscriptions at once.
-//
-// For code that handle more than one subscription, a scope can be used to conveniently
-// unsubscribe all of them with a single call. The example demonstrates a typical use in a
-// larger program.
-//
-// The zero value is ready to use.
-type SubscriptionScope struct {
- mu sync.Mutex
- subs map[*scopeSub]struct{}
- closed bool
-}
-
-type scopeSub struct {
- sc *SubscriptionScope
- s Subscription
-}
-
-// Track starts tracking a subscription. If the scope is closed, Track returns nil. The
-// returned subscription is a wrapper. Unsubscribing the wrapper removes it from the
-// scope.
-func (sc *SubscriptionScope) Track(s Subscription) Subscription {
- sc.mu.Lock()
- defer sc.mu.Unlock()
- if sc.closed {
- return nil
- }
- if sc.subs == nil {
- sc.subs = make(map[*scopeSub]struct{})
- }
- ss := &scopeSub{sc, s}
- sc.subs[ss] = struct{}{}
- return ss
-}
-
-// Close calls Unsubscribe on all tracked subscriptions and prevents further additions to
-// the tracked set. Calls to Track after Close return nil.
-func (sc *SubscriptionScope) Close() {
- sc.mu.Lock()
- defer sc.mu.Unlock()
- if sc.closed {
- return
- }
- sc.closed = true
- for s := range sc.subs {
- s.s.Unsubscribe()
- }
- sc.subs = nil
-}
-
-// Count returns the number of tracked subscriptions.
-// It is meant to be used for debugging.
-func (sc *SubscriptionScope) Count() int {
- sc.mu.Lock()
- defer sc.mu.Unlock()
- return len(sc.subs)
-}
-
-func (s *scopeSub) Unsubscribe() {
- s.s.Unsubscribe()
- s.sc.mu.Lock()
- defer s.sc.mu.Unlock()
- delete(s.sc.subs, s)
-}
-
-func (s *scopeSub) Err() <-chan error {
- return s.s.Err()
-}
diff --git a/event/subscription_test.go b/event/subscription_test.go
deleted file mode 100644
index 743d0bf67d..0000000000
--- a/event/subscription_test.go
+++ /dev/null
@@ -1,180 +0,0 @@
-// Copyright 2016 The go-ethereum Authors
-// This file is part of the go-ethereum library.
-//
-// The go-ethereum library is free software: you can redistribute it and/or modify
-// it under the terms of the GNU Lesser General Public License as published by
-// the Free Software Foundation, either version 3 of the License, or
-// (at your option) any later version.
-//
-// The go-ethereum library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
-// GNU Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public License
-// along with the go-ethereum library. If not, see .
-
-package event
-
-import (
- "context"
- "errors"
- "fmt"
- "reflect"
- "testing"
- "time"
-)
-
-var errInts = errors.New("error in subscribeInts")
-
-func subscribeInts(max, fail int, c chan<- int) Subscription {
- return NewSubscription(func(quit <-chan struct{}) error {
- for i := 0; i < max; i++ {
- if i >= fail {
- return errInts
- }
- select {
- case c <- i:
- case <-quit:
- return nil
- }
- }
- return nil
- })
-}
-
-func TestNewSubscriptionError(t *testing.T) {
- t.Parallel()
-
- channel := make(chan int)
- sub := subscribeInts(10, 2, channel)
-loop:
- for want := 0; want < 10; want++ {
- select {
- case got := <-channel:
- if got != want {
- t.Fatalf("wrong int %d, want %d", got, want)
- }
- case err := <-sub.Err():
- if err != errInts {
- t.Fatalf("wrong error: got %q, want %q", err, errInts)
- }
- if want != 2 {
- t.Fatalf("got errInts at int %d, should be received at 2", want)
- }
- break loop
- }
- }
- sub.Unsubscribe()
-
- err, ok := <-sub.Err()
- if err != nil {
- t.Fatal("got non-nil error after Unsubscribe")
- }
- if ok {
- t.Fatal("channel still open after Unsubscribe")
- }
-}
-
-func TestResubscribe(t *testing.T) {
- t.Parallel()
-
- var i int
- nfails := 6
- sub := Resubscribe(100*time.Millisecond, func(ctx context.Context) (Subscription, error) {
- // fmt.Printf("call #%d @ %v\n", i, time.Now())
- i++
- if i == 2 {
- // Delay the second failure a bit to reset the resubscribe interval.
- time.Sleep(200 * time.Millisecond)
- }
- if i < nfails {
- return nil, errors.New("oops")
- }
- sub := NewSubscription(func(unsubscribed <-chan struct{}) error { return nil })
- return sub, nil
- })
-
- <-sub.Err()
- if i != nfails {
- t.Fatalf("resubscribe function called %d times, want %d times", i, nfails)
- }
-}
-
-func TestResubscribeAbort(t *testing.T) {
- t.Parallel()
-
- done := make(chan error, 1)
- sub := Resubscribe(0, func(ctx context.Context) (Subscription, error) {
- select {
- case <-ctx.Done():
- done <- nil
- case <-time.After(2 * time.Second):
- done <- errors.New("context given to resubscribe function not canceled within 2s")
- }
- return nil, nil
- })
-
- sub.Unsubscribe()
- if err := <-done; err != nil {
- t.Fatal(err)
- }
-}
-
-func TestResubscribeWithErrorHandler(t *testing.T) {
- t.Parallel()
-
- var i int
- nfails := 6
- subErrs := make([]string, 0)
- sub := ResubscribeErr(100*time.Millisecond, func(ctx context.Context, lastErr error) (Subscription, error) {
- i++
- var lastErrVal string
- if lastErr != nil {
- lastErrVal = lastErr.Error()
- }
- subErrs = append(subErrs, lastErrVal)
- sub := NewSubscription(func(unsubscribed <-chan struct{}) error {
- if i < nfails {
- return fmt.Errorf("err-%v", i)
- } else {
- return nil
- }
- })
- return sub, nil
- })
-
- <-sub.Err()
- if i != nfails {
- t.Fatalf("resubscribe function called %d times, want %d times", i, nfails)
- }
-
- expectedSubErrs := []string{"", "err-1", "err-2", "err-3", "err-4", "err-5"}
- if !reflect.DeepEqual(subErrs, expectedSubErrs) {
- t.Fatalf("unexpected subscription errors %v, want %v", subErrs, expectedSubErrs)
- }
-}
-
-func TestResubscribeWithCompletedSubscription(t *testing.T) {
- t.Parallel()
-
- quitProducerAck := make(chan struct{})
- quitProducer := make(chan struct{})
-
- sub := ResubscribeErr(100*time.Millisecond, func(ctx context.Context, lastErr error) (Subscription, error) {
- return NewSubscription(func(unsubscribed <-chan struct{}) error {
- select {
- case <-quitProducer:
- quitProducerAck <- struct{}{}
- return nil
- case <-unsubscribed:
- return nil
- }
- }), nil
- })
-
- // Ensure producer has started and exited before Unsubscribe
- close(quitProducer)
- <-quitProducerAck
- sub.Unsubscribe()
-}