pipo tx parse

pkg/shreder/client.go

@@ -70,7 +70,7 @@ func NewShrederClient(
 
 	poolSize := runtime.NumCPU()*2 + 2
 	logger.Info("creating shreder client", "url", url, "pool_size", poolSize)
-	pool, err := ants.NewPool(poolSize, ants.WithNonblocking(false))
+	pool, err := ants.NewPool(poolSize, ants.WithNonblocking(true))
 	if err != nil {
 		return nil, func() {}, err
 	}
@@ -113,7 +113,42 @@ 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
+			}
+		}
+
+		entries := response.Entries
+
+		err = c.pool.Submit(func() {
+			ParseEntries(slot, entries, c.tableLoader, txCh, c.enableParseStats)
+		})
+		if err != nil {
+			return err
+		}
+	}
+}
+
+func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignal) error {
 	stream, err := c.client.SubscribeTransactions(ctx)
 	if err != nil {
 		return err
@@ -149,23 +184,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:
-			}
+			ParseTransaction(txData, c.tableLoader, txCh, c.enableParseStats)
 		})
 		if err != nil {
-			return err
+			logger.Error("failed to submit transaction: ", "err", err)
 		}
+
 	}
 }