Add subscribeEvents and subcriptionEvent

blockchain/blockchain.go

@@ -44,7 +44,7 @@ type Reader interface {
 
 	BlockCommitmentsByNumber(blockNumber uint64) (*core.BlockCommitments, error)
 
-	EventFilter(from *felt.Felt, keys [][]felt.Felt) (*EventFilter, error)
+	EventFilter(from *felt.Felt, keys [][]felt.Felt) (EventFilterer, error)
 
 	Pending() (Pending, error)
 
@@ -854,7 +854,7 @@ func (b *Blockchain) StateAtBlockHash(blockHash *felt.Felt) (core.StateReader, S
 }
 
 // EventFilter returns an EventFilter object that is tied to a snapshot of the blockchain
-func (b *Blockchain) EventFilter(from *felt.Felt, keys [][]felt.Felt) (*EventFilter, error) {
+func (b *Blockchain) EventFilter(from *felt.Felt, keys [][]felt.Felt) (EventFilterer, error) {
 	b.listener.OnRead("EventFilter")
 	txn, err := b.database.NewTransaction(false)
 	if err != nil {

blockchain/event_filter.go

@@ -4,6 +4,7 @@ import (
 	"encoding/binary"
 	"errors"
 	"fmt"
+	"io"
 	"math"
 
 	"github.com/NethermindEth/juno/core"
@@ -14,6 +15,16 @@ import (
 
 var errChunkSizeReached = errors.New("chunk size reached")
 
+//go:generate mockgen -destination=../mocks/mock_event_filterer.go -package=mocks github.com/NethermindEth/juno/blockchain EventFilterer
+type EventFilterer interface {
+	io.Closer
+
+	Events(cToken *ContinuationToken, chunkSize uint64) ([]*FilteredEvent, *ContinuationToken, error)
+	SetRangeEndBlockByNumber(filterRange EventFilterRange, blockNumber uint64) error
+	SetRangeEndBlockByHash(filterRange EventFilterRange, blockHash *felt.Felt) error
+	WithLimit(limit uint) *EventFilter
+}
+
 type EventFilter struct {
 	txn             db.Transaction
 	fromBlock       uint64

rpc/events.go

@@ -44,6 +44,10 @@ type EventsChunk struct {
 	ContinuationToken string          `json:"continuation_token,omitempty"`
 }
 
+type SubscriptionID struct {
+	ID uint64 `json:"subscription_id"`
+}
+
 /****************************************************
 		Events Handlers
 *****************************************************/
@@ -74,7 +78,7 @@ func (h *Handler) SubscribeNewHeads(ctx context.Context) (uint64, *jsonrpc.Error
 			case <-subscriptionCtx.Done():
 				return
 			case header := <-headerSub.Recv():
-				resp, err := json.Marshal(jsonrpc.Request{
+				resp, err := json.Marshal(SubscriptionResponse{
 					Version: "2.0",
 					Method:  "juno_subscribeNewHeads",
 					Params: map[string]any{
@@ -191,7 +195,7 @@ func (h *Handler) unsubscribe(sub *subscription, id uint64) {
 	h.mu.Unlock()
 }
 
-func setEventFilterRange(filter *blockchain.EventFilter, fromID, toID *BlockID, latestHeight uint64) error {
+func setEventFilterRange(filter blockchain.EventFilterer, fromID, toID *BlockID, latestHeight uint64) error {
 	set := func(filterRange blockchain.EventFilterRange, id *BlockID) error {
 		if id == nil {
 			return nil

rpc/handlers.go

@@ -5,6 +5,7 @@ import (
 	"crypto/rand"
 	"encoding/binary"
 	"encoding/json"
+	"fmt"
 	"log"
 	"math"
 	"strings"
@@ -65,12 +66,15 @@ var (
 	ErrUnsupportedTxVersion            = &jsonrpc.Error{Code: 61, Message: "the transaction version is not supported"}
 	ErrUnsupportedContractClassVersion = &jsonrpc.Error{Code: 62, Message: "the contract class version is not supported"}
 	ErrUnexpectedError                 = &jsonrpc.Error{Code: 63, Message: "An unexpected error occurred"}
+	ErrTooManyBlocksBack               = &jsonrpc.Error{Code: 68, Message: fmt.Sprintf("Cannot go back more than %v blocks", maxBlocksBack)}
+	ErrCallOnPending                   = &jsonrpc.Error{Code: 69, Message: "This method does not support being called on the pending block"}
 
 	// These errors can be only be returned by Juno-specific methods.
 	ErrSubscriptionNotFound = &jsonrpc.Error{Code: 100, Message: "Subscription not found"}
 )
 
 const (
+	maxBlocksBack      = 1024
 	maxEventChunkSize  = 10240
 	maxEventFilterKeys = 1024
 	traceCacheSize     = 128
@@ -334,6 +338,11 @@ func (h *Handler) Methods() ([]jsonrpc.Method, string) { //nolint: funlen
 			Name:    "starknet_specVersion",
 			Handler: h.SpecVersion,
 		},
+		{
+			Name:    "starknet_subscribeEvents",
+			Params:  []jsonrpc.Parameter{{Name: "from_address", Optional: true}, {Name: "keys", Optional: true}, {Name: "block", Optional: true}},
+			Handler: h.SubscribeEvents,
+		},
 		{
 			Name:    "juno_subscribeNewHeads",
 			Handler: h.SubscribeNewHeads,