wip

Author: IronGauntlets

rpc/subscriptions.go

@@ -3,6 +3,8 @@ package rpc
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"reflect"
 	"time"
 
 	"github.com/NethermindEth/juno/blockchain"
@@ -77,8 +79,8 @@ func (h *Handler) SubscribeEvents(ctx context.Context, fromAddr *felt.Felt, keys
 		// old blocks.
 		var wg conc.WaitGroup
 		wg.Go(func() {
-			// Stores the transaction hash of the event
-			eventsPreviouslySent := make(map[felt.Felt]struct{})
+			// Stores the transaction hash -> number of events
+			eventsPreviouslySent := make([]*blockchain.FilteredEvent, 0)
 
 			for {
 				select {
@@ -98,10 +100,34 @@ func (h *Handler) SubscribeEvents(ctx context.Context, fromAddr *felt.Felt, keys
 						return
 					}
 
-					for _, t := range b.Transactions {
-						delete(eventsPreviouslySent, *t.Hash())
+					fmt.Println("size of slice before", len(eventsPreviouslySent))
+					for i, r := range b.Receipts {
+						for _, e := range r.Events {
+							fe := &blockchain.FilteredEvent{
+								Event:           e,
+								BlockNumber:     header.Number,
+								BlockHash:       header.Hash,
+								TransactionHash: b.Transactions[i].Hash(),
+							}
+
+							var deleteI int
+							var duplicateFound bool
+							for j, dupE := range eventsPreviouslySent {
+								if reflect.DeepEqual(fe, dupE) {
+									duplicateFound = true
+									deleteI = j
+									break
+								}
+							}
+
+							if duplicateFound {
+								eventsPreviouslySent = append(eventsPreviouslySent[:deleteI], eventsPreviouslySent[deleteI+1:]...)
+							}
+						}
 					}
+					fmt.Println("size of slice after", len(eventsPreviouslySent))
 				case pending := <-pendingSub.Recv():
+					fmt.Println("Found pending block", len(pending.Transactions))
 					h.processEvents(subscriptionCtx, w, id, pending.Number, pending.Number, fromAddr, keys, eventsPreviouslySent)
 				}
 			}
@@ -271,7 +297,7 @@ func (h *Handler) SubscribeTransactionStatus(ctx context.Context, txHash felt.Fe
 }
 
 func (h *Handler) processEvents(ctx context.Context, w jsonrpc.Conn, id, from, to uint64, fromAddr *felt.Felt,
-	keys [][]felt.Felt, eventsPreviouslySent map[felt.Felt]struct{},
+	keys [][]felt.Felt, eventsPreviouslySent []*blockchain.FilteredEvent,
 ) {
 	filter, err := h.bcReader.EventFilter(fromAddr, keys)
 	if err != nil {
@@ -314,18 +340,19 @@ func (h *Handler) processEvents(ctx context.Context, w jsonrpc.Conn, id, from, t
 }
 
 func sendEvents(ctx context.Context, w jsonrpc.Conn, events []*blockchain.FilteredEvent,
-	eventsPreviouslySent map[felt.Felt]struct{}, id uint64,
+	eventsPreviouslySent []*blockchain.FilteredEvent, id uint64,
 ) error {
+eventsLoop:
 	for _, event := range events {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
 		default:
-			if eventsPreviouslySent != nil {
-				if _, exists := eventsPreviouslySent[*event.TransactionHash]; exists {
-					continue
+			for _, prevEvent := range eventsPreviouslySent {
+				if reflect.DeepEqual(event, prevEvent) {
+					continue eventsLoop
 				}
-				eventsPreviouslySent[*event.TransactionHash] = struct{}{}
+				eventsPreviouslySent = append(eventsPreviouslySent, event)
 			}
 
 			emittedEvent := &EmittedEvent{

rpc/subscriptions_test.go

@@ -7,6 +7,7 @@ import (
 	"io"
 	"net"
 	"net/http/httptest"
+	"reflect"
 	"testing"
 	"time"
 
@@ -1031,3 +1032,55 @@ func marshalSubEventsResp(e *EmittedEvent, id uint64) ([]byte, error) {
 		},
 	})
 }
+
+func TestEventEquality(t *testing.T) {
+	e1 := &Event{
+		From: new(felt.Felt).SetUint64(1),
+		Keys: []*felt.Felt{new(felt.Felt).SetUint64(2), new(felt.Felt).SetUint64(3)},
+		Data: []*felt.Felt{new(felt.Felt).SetUint64(4), new(felt.Felt).SetUint64(3)},
+	}
+
+	e2 := &Event{
+		From: new(felt.Felt).SetUint64(1),
+		Keys: []*felt.Felt{new(felt.Felt).SetUint64(2), new(felt.Felt).SetUint64(3)},
+		Data: []*felt.Felt{new(felt.Felt).SetUint64(4), new(felt.Felt).SetUint64(3)},
+	}
+
+	assert.True(t, reflect.DeepEqual(e1, e2))
+
+	e3 := &core.Event{
+		From: new(felt.Felt).SetUint64(1),
+		Keys: []*felt.Felt{new(felt.Felt).SetUint64(2), new(felt.Felt).SetUint64(3)},
+		Data: []*felt.Felt{new(felt.Felt).SetUint64(4), new(felt.Felt).SetUint64(3)},
+	}
+
+	e4 := &core.Event{
+		From: new(felt.Felt).SetUint64(1),
+		Keys: []*felt.Felt{new(felt.Felt).SetUint64(2), new(felt.Felt).SetUint64(3)},
+		Data: []*felt.Felt{new(felt.Felt).SetUint64(4), new(felt.Felt).SetUint64(3)},
+	}
+
+	assert.True(t, reflect.DeepEqual(e3, e4))
+
+	bn1 := uint64(10)
+	ee1 := EmittedEvent{
+		Event:           e1,
+		BlockNumber:     &bn1,
+		BlockHash:       new(felt.Felt).SetBytes([]byte("b hash")),
+		TransactionHash: new(felt.Felt).SetBytes([]byte("tx hash")),
+	}
+
+	bn2 := uint64(10)
+	ee2 := EmittedEvent{
+		Event:           e2,
+		BlockNumber:     &bn2,
+		BlockHash:       new(felt.Felt).SetBytes([]byte("b hash")),
+		TransactionHash: new(felt.Felt).SetBytes([]byte("tx hash")),
+	}
+
+	assert.True(t, reflect.DeepEqual(ee1, ee2))
+	// assert.True(t, *ee1.Event.Data == *ee2.Event.Data)
+	assert.True(t, *ee1.BlockNumber == *ee2.BlockNumber)
+	assert.True(t, *ee1.BlockHash == *ee2.BlockHash)
+	assert.True(t, *ee1.TransactionHash == *ee2.TransactionHash)
+}