add test to reproduce panic (#95)

client_test.go

@@ -465,3 +465,88 @@ func TestConcurrentPublishSubscribe(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestConcurrentPublishSubscribeDisconnect(t *testing.T) {
+	// The purpose of this test is to try to catch possible race conditions
+	// that can happen when client is disconnected while receiving messages.
+	const (
+		numMessages        = 1000
+		numResubscritpions = 100
+	)
+
+	producer := NewJsonClient("ws://localhost:8000/connection/websocket?cf_protocol_version=v2", Config{})
+	defer producer.Close()
+
+	if err := producer.Connect(); err != nil {
+		t.Fatalf("error on connect: %v", err)
+	}
+
+	errChan := make(chan error)
+	defer close(errChan)
+	go func() {
+		for i := 0; i < numMessages; i++ {
+			msg := []byte(`{"unique":"` + randString(6) + strconv.FormatInt(time.Now().UnixNano(), 10) + `"}`)
+			_, err := producer.Publish(context.Background(), "test_concurrent", msg)
+			if err != nil {
+				errChan <- fmt.Errorf("error on publish: %v", err)
+				return
+			}
+		}
+		errChan <- nil
+	}()
+
+	go func() {
+		for i := 0; i < numResubscritpions; i++ {
+			consumer := NewJsonClient("ws://localhost:8000/connection/websocket?cf_protocol_version=v2", Config{})
+			if err := consumer.Connect(); err != nil {
+				errChan <- fmt.Errorf("error on connect: %v", err)
+				return
+			}
+
+			handler := &testSubscriptionHandler{
+				onPublication: func(e PublicationEvent) {
+					// We just want the callback queue to do its jobs.
+					time.Sleep(time.Microsecond)
+				},
+			}
+			sub, err := consumer.NewSubscription("test_concurrent")
+			if err != nil {
+				errChan <- fmt.Errorf("error on new subscription: %v (%d)", err, i)
+				return
+			}
+			sub.OnSubscribed(handler.OnSubscribe)
+			sub.OnPublication(handler.OnPublication)
+			if err := sub.Subscribe(); err != nil {
+				errChan <- fmt.Errorf("error on subscribe: %v (%d)", err, i)
+				return
+			}
+			sub2, err := consumer.NewSubscription("something_else")
+			if err != nil {
+				errChan <- fmt.Errorf("error on new subscription: %v (%d)", err, i)
+				return
+			}
+			sub2.OnSubscribed(handler.OnSubscribe)
+			sub2.OnPublication(handler.OnPublication)
+			if err := sub2.Subscribe(); err != nil {
+				errChan <- fmt.Errorf("error on subscribe: %v (%d)", err, i)
+				return
+			}
+			// Simulate random disconnects.
+			go func(cl *Client) {
+				time.Sleep(time.Duration(rand.Int63n(150)) * time.Millisecond)
+				cl.Close()
+			}(consumer)
+		}
+		errChan <- nil
+	}()
+
+	var err error
+	for i := 0; i < 2; i++ {
+		if e := <-errChan; e != nil {
+			err = e
+		}
+	}
+	if err != nil {
+		t.Fatal(err)
+	}
+}