entries custom filter and parse

pkg/shreder/client.go

@@ -8,6 +8,7 @@ import (
 	"runtime"
 	"time"
 
+	"github.com/gagliardetto/solana-go"
 	"github.com/gagliardetto/solana-go/rpc"
 	"github.com/panjf2000/ants/v2"
 	"google.golang.org/grpc"
@@ -23,6 +24,10 @@ type Client struct {
 	tableLoader  *AddressTables
 	subscription map[string]*SubscribeRequestFilterTransactions
 
+	entriesFilter map[string]FilterParams
+
+	parser map[solana.PublicKey]Handler
+
 	pool *ants.Pool
 
 	lastSlot     uint64
@@ -33,6 +38,8 @@ type ClientOpts struct {
 	blockStats      bool
 	showTableLoaded bool
 	logParseStats   bool
+
+	parser map[solana.PublicKey]Handler
 }
 
 type ClientOption func(*ClientOpts)
@@ -43,6 +50,12 @@ func ShowTableLoaded(enable bool) ClientOption {
 	}
 }
 
+func WithCustomParsers(parsers map[solana.PublicKey]Handler) ClientOption {
+	return func(opts *ClientOpts) {
+		opts.parser = parsers
+	}
+}
+
 func BlocksStats(enable bool) ClientOption {
 	return func(opts *ClientOpts) {
 		opts.blockStats = enable
@@ -82,16 +95,33 @@ func NewShrederClient(
 		blockStats:      false,
 		showTableLoaded: true,
 		logParseStats:   false,
+		parser:          registered,
 	}
 	for _, option := range options {
 		option(o)
 	}
+	filterParams := make(map[string]FilterParams)
+	for name, params := range subscription {
+		filterParams[name] = FilterParams{
+			Exclude: parseAccountArray(params.AccountExclude),
+			Require: parseAccountArray(params.AccountRequired),
+			Include: parseAccountArray(params.AccountInclude),
+		}
+	}
+	if len(filterParams) == 0 {
+		filterParams["default"] = FilterParams{
+			Include: defaultFilterAccount,
+		}
+	}
+
 	s := &Client{
-		conn:         conn,
-		client:       NewShrederServiceClient(conn),
-		subscription: subscription,
-		tableLoader:  NewAddressTables(rpcClient, o.showTableLoaded),
-		pool:         pool,
+		conn:          conn,
+		client:        NewShrederServiceClient(conn),
+		subscription:  subscription,
+		entriesFilter: filterParams,
+		parser:        o.parser,
+		tableLoader:   NewAddressTables(rpcClient, o.showTableLoaded),
+		pool:          pool,
 
 		enableBlockStats: o.blockStats,
 		enableParseStats: o.logParseStats,
@@ -142,7 +172,16 @@ func (c *Client) ReadEntriesSync(ctx context.Context, txCh chan<- TxSignal) erro
 		}
 
 		err = c.pool.Submit(func() {
-			ParseTransactionForEntries(ctx, slot, bytes.NewReader(response.Entries), c.tableLoader, txCh)
+			err := entriesToVersionedTransaction(slot, bytes.NewReader(response.Entries), func(versioned VersionedTransaction) {
+				// filter out vote transactions
+				if FilterTransactionForEntriesWithFilter(versioned, c.entriesFilter) {
+					return
+				}
+				go ParseTransactionWithHandler(ctx, versioned, c.tableLoader, txCh, c.parser)
+			})
+			if err != nil {
+				logger.Debug("txparser: failed to parse entries", "error", err)
+			}
 		})
 		if err != nil && errors.Is(err, ants.ErrPoolOverload) {
 			logger.Warn("task pool is full")
@@ -187,10 +226,15 @@ func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignal) error {
 			}
 		}
 
-		txData := response.Transaction
+		// txData := response.Transaction
 
 		err := c.pool.Submit(func() {
-			ParseTransactionForSubscribe(ctx, txData, c.tableLoader, txCh, nil)
+			versioned, err := toVersionedTransaction(response.Transaction)
+			if err != nil {
+				logger.Debug("txparser: failed to convert to versioned transaction", "error", err)
+				return
+			}
+			ParseTransactionWithHandler(ctx, versioned, c.tableLoader, txCh, c.parser)
 		})
 		if err != nil && errors.Is(err, ants.ErrPoolOverload) {
 			logger.Warn("task pool is full")
@@ -202,3 +246,11 @@ func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignal) error {
 
 	return err
 }
+
+func parseAccountArray(accountArray []string) []solana.PublicKey {
+	var result []solana.PublicKey
+	for _, acc := range accountArray {
+		result = append(result, solana.MustPublicKeyFromBase58(acc))
+	}
+	return result
+}