split program source file

pkg/shreder/client.go

@@ -1,7 +1,9 @@
 package shreder
 
 import (
+	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"runtime"
 	"time"
@@ -47,6 +49,8 @@ func BlocksStats(enable bool) ClientOption {
 	}
 }
 
+// LogParsedStats enables logging of parsed transaction statistics.
+// Deprecated: do not use.
 func LogParsedStats(enable bool) ClientOption {
 	return func(opts *ClientOpts) {
 		opts.logParseStats = enable
@@ -113,7 +117,40 @@ func (c *Client) Wait() {
 	logger.Debug("shreder client stopped")
 }
 
-func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignalBatch) error {
+func (c *Client) ReadEntriesSync(ctx context.Context, txCh chan<- TxSignal) error {
+	stream, err := c.client.SubscribeEntries(ctx, &SubscribeEntriesRequest{})
+	if err != nil {
+		return err
+	}
+
+	logger.Debug("reading entries from shreder client")
+	for {
+		response, err := stream.Recv()
+		if err != nil {
+			return err
+		}
+		slot := response.Slot
+		if c.enableBlockStats {
+			now := time.Now()
+			if c.lastSlotTime.IsZero() || slot > c.lastSlot {
+				if !c.lastSlotTime.IsZero() {
+					logger.Info("block processed", "running", c.pool.Running(), "slot", slot, "prev_slot", c.lastSlot, "delta_ms", now.Sub(c.lastSlotTime).Milliseconds())
+				}
+				c.lastSlot = slot
+				c.lastSlotTime = now
+			}
+		}
+
+		err = c.pool.Submit(func() {
+			ParseTransactionForEntries(ctx, slot, bytes.NewReader(response.Entries), c.tableLoader, txCh)
+		})
+		if err != nil && errors.Is(err, ants.ErrPoolOverload) {
+			logger.Warn("task pool is full")
+		}
+	}
+}
+
+func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignal) error {
 	stream, err := c.client.SubscribeTransactions(ctx)
 	if err != nil {
 		return err
@@ -152,24 +189,11 @@ func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignalBatch) error
 
 		txData := response.Transaction
 
-		err = c.pool.Submit(func() {
-			txBatch := ParseTransaction(txData, c.tableLoader, c.enableParseStats)
-			if len(txBatch) == 0 {
-				return
-			}
-
-			for _, tx := range txBatch {
-				tx.Source = "shreder"
-			}
-
-			select {
-			case <-ctx.Done():
-				return
-			case txCh <- txBatch:
-			}
+		err := c.pool.Submit(func() {
+			ParseTransactionForSubscribe(ctx, txData, c.tableLoader, txCh, nil)
 		})
-		if err != nil {
-			break
+		if err != nil && errors.Is(err, ants.ErrPoolOverload) {
+			logger.Warn("task pool is full")
 		}
 	}