From 525a0e67fe8b9e786ba4a4f3a3615a50eca7799c Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=C5=81ukasz=20Magiera?= <magik6k@gmail.com>
Date: Sat, 22 Jun 2019 12:05:03 +0200
Subject: [PATCH 1/3] fix close deadlock and Sub type error

---
 basic.go      | 28 +++++++++++++++++++++++-----
 basic_test.go | 21 +++++++++++++++++++++
 2 files changed, 44 insertions(+), 5 deletions(-)

diff --git a/basic.go b/basic.go
index ca102b5..2fe3798 100644
--- a/basic.go
+++ b/basic.go
@@ -104,9 +104,21 @@ func (s *sub) Out() <-chan interface{} {
 }
 
 func (s *sub) Close() error {
-	close(s.ch)
+	stop := make(chan struct{})
+	go func() {
+		for {
+			select {
+			case <-s.ch:
+			case <-stop:
+				close(s.ch)
+				return
+			}
+		}
+	}()
+
 	for _, n := range s.nodes {
 		n.lk.Lock()
+
 		for i := 0; i < len(n.sinks); i++ {
 			if n.sinks[i] == s.ch {
 				n.sinks[i], n.sinks[len(n.sinks)-1] = n.sinks[len(n.sinks)-1], nil
@@ -114,12 +126,16 @@ func (s *sub) Close() error {
 				break
 			}
 		}
+
 		tryDrop := len(n.sinks) == 0 && atomic.LoadInt32(&n.nEmitters) == 0
+
 		n.lk.Unlock()
+
 		if tryDrop {
 			s.dropper(n.typ)
 		}
 	}
+	close(stop)
 	return nil
 }
 
@@ -148,12 +164,14 @@ func (b *basicBus) Subscribe(evtTypes interface{}, opts ...event.SubscriptionOpt
 		dropper: b.tryDropNode,
 	}
 
-	for i, etyp := range types {
-		typ := reflect.TypeOf(etyp)
-
-		if typ.Kind() != reflect.Ptr {
+	for _, etyp := range types {
+		if reflect.TypeOf(etyp).Kind() != reflect.Ptr {
 			return nil, errors.New("subscribe called with non-pointer type")
 		}
+	}
+
+	for i, etyp := range types {
+		typ := reflect.TypeOf(etyp)
 
 		err = b.withNode(typ.Elem(), func(n *node) {
 			n.sinks = append(n.sinks, out.ch)
diff --git a/basic_test.go b/basic_test.go
index fc23e61..62b84eb 100644
--- a/basic_test.go
+++ b/basic_test.go
@@ -297,6 +297,27 @@ func TestStateful(t *testing.T) {
 	}
 }
 
+func TestCloseBlocking(t *testing.T) {
+	bus := NewBus()
+	em, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sub, err := bus.Subscribe(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	go func() {
+		em.Emit(EventB(159))
+	}()
+
+	time.Sleep(10 * time.Millisecond) // make sure that emit is blocked
+
+	sub.Close() // cancel sub
+}
+
 func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
 	if race.WithRace() && subs+emits > 5000 {
 		t.SkipNow()

From 3fd76c7eacaa0f130ffe9f6963183e164524ccb1 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=C5=81ukasz=20Magiera?= <magik6k@gmail.com>
Date: Sat, 22 Jun 2019 15:14:04 +0200
Subject: [PATCH 2/3] Add test for #10

---
 basic_test.go | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/basic_test.go b/basic_test.go
index 62b84eb..aaad88a 100644
--- a/basic_test.go
+++ b/basic_test.go
@@ -318,6 +318,21 @@ func TestCloseBlocking(t *testing.T) {
 	sub.Close() // cancel sub
 }
 
+func TestSubFailFully(t *testing.T) {
+	bus := NewBus()
+	em, err := bus.Emitter(new(EventB))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	_, err = bus.Subscribe([]interface{}{new(EventB), 5})
+	if err == nil || err.Error() != "subscribe called with non-pointer type" {
+		t.Fatal(err)
+	}
+
+	em.Emit(EventB(159)) // will hang if sub doesn't fail properly
+}
+
 func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
 	if race.WithRace() && subs+emits > 5000 {
 		t.SkipNow()

From 454cbe549777ceb63b4e41bd9cd3790efdb89e2c Mon Sep 17 00:00:00 2001
From: Jakub Sztandera <kubuxu@protonmail.ch>
Date: Sat, 22 Jun 2019 15:45:09 +0200
Subject: [PATCH 3/3] Make the test fail in reasonable time

License: MIT
Signed-off-by: Jakub Sztandera <kubuxu@protonmail.ch>
---
 basic_test.go | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/basic_test.go b/basic_test.go
index aaad88a..49cc4d7 100644
--- a/basic_test.go
+++ b/basic_test.go
@@ -318,6 +318,11 @@ func TestCloseBlocking(t *testing.T) {
 	sub.Close() // cancel sub
 }
 
+func panicOnTimeout(d time.Duration) {
+	<-time.After(d)
+	panic("timeout reached")
+}
+
 func TestSubFailFully(t *testing.T) {
 	bus := NewBus()
 	em, err := bus.Emitter(new(EventB))
@@ -330,6 +335,8 @@ func TestSubFailFully(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	go panicOnTimeout(5 * time.Second)
+
 	em.Emit(EventB(159)) // will hang if sub doesn't fail properly
 }