Merge pull request #37 from kaleido-io/reply-chl-depth

Move 10 depth to correct channel, for spurious ack processing

Author: nguyer

internal/confirmations/confirmations.go

@@ -150,7 +150,7 @@ func (pi *pendingItem) getKey() string {
 	case pendingTypeEvent:
 		// For events they are identified by their hash, blockNumber, transactionIndex and logIndex
 		// If any of those change, it's a new new event - and as such we should get informed of it separately by the blockchain connector.
-		return fmt.Sprintf("Event:l=%s,th=%s,bh=%s,bn=%d,ti=%d,li=%d", pi.listenerID, pi.transactionHash, pi.blockHash, pi.blockNumber, pi.transactionIndex, pi.logIndex)
+		return fmt.Sprintf("Event:%.12d/%.6d/%.6d,l=%s,th=%s,bh=%s", pi.blockNumber, pi.transactionIndex, pi.logIndex, pi.listenerID, pi.transactionHash, pi.blockHash)
 	case pendingTypeTransaction:
 		// For transactions, it's simply the transaction hash that identifies it. It can go into any block
 		return pendingKeyForTX(pi.transactionHash)

internal/ws/wsconn.go

@@ -39,7 +39,6 @@ type webSocketConnection struct {
 	topics    map[string]*webSocketTopic
 	broadcast chan interface{}
 	newTopic  chan bool
-	receive   chan *WebSocketCommandMessageOrError
 	closing   chan struct{}
 }
 
@@ -66,7 +65,6 @@ func newConnection(bgCtx context.Context, server *webSocketServer, conn *ws.Conn
 		newTopic:  make(chan bool),
 		topics:    make(map[string]*webSocketTopic),
 		broadcast: make(chan interface{}),
-		receive:   make(chan *WebSocketCommandMessageOrError, 10),
 		closing:   make(chan struct{}),
 	}
 	go wsc.listen()

internal/ws/wsserver.go

@@ -132,7 +132,7 @@ func (s *webSocketServer) getTopic(topic string) *webSocketTopic {
 			topic:            topic,
 			senderChannel:    make(chan interface{}),
 			broadcastChannel: make(chan interface{}),
-			receiverChannel:  make(chan *WebSocketCommandMessageOrError, 1),
+			receiverChannel:  make(chan *WebSocketCommandMessageOrError, 10),
 		}
 		s.topics[topic] = t
 		s.topicMap[topic] = make(map[string]*webSocketConnection)

internal/ws/wsserver_test.go

@@ -189,6 +189,10 @@ func TestSpuriousAckProcessing(t *testing.T) {
 	c, _, err := ws.DefaultDialer.Dial(u.String(), nil)
 	assert.NoError(err)
 
+	// Drop depth to 1 for spurious ack processing
+	topic := w.getTopic("mytopic")
+	topic.receiverChannel = make(chan *WebSocketCommandMessageOrError, 1)
+
 	c.WriteJSON(&WebSocketCommandMessage{
 		Type:  "ack",
 		Topic: "mytopic",
@@ -217,6 +221,10 @@ func TestSpuriousNackProcessing(t *testing.T) {
 	c, _, err := ws.DefaultDialer.Dial(u.String(), nil)
 	assert.NoError(err)
 
+	// Drop depth to 1 for spurious ack processing
+	topic := w.getTopic("mytopic")
+	topic.receiverChannel = make(chan *WebSocketCommandMessageOrError, 1)
+
 	c.WriteJSON(&WebSocketCommandMessage{
 		Type:  "ack",
 		Topic: "mytopic",