chore: add shrederClient exaple

cmd/shreder/main.go

@@ -1,7 +1,76 @@
 package main
 
-import "fmt"
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/samlior/libsam/pkg/logger"
+	"github.com/samlior/libsam/pkg/shreder"
+	"github.com/samlior/libsam/pkg/types"
+	"github.com/samlior/libsam/third_party/shreder_protos"
+)
 
 func main() {
-	fmt.Println("Hello, World!")
+	url := os.Getenv("URL")
+	if url == "" {
+		panic("URL is not set")
+	}
+
+	logger := logger.NewEmptyLogger()
+	shrederClient, cleanup, err := shreder.NewShrederClient(
+		logger,
+		url,
+		map[string]*shreder_protos.SubscribeRequestFilterTransactions{
+			"pumpfunamm": {
+				AccountRequired: []string{
+					"pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA",
+				},
+			},
+			"photon": {
+				AccountRequired: []string{
+					"BSfD6SHZigAfDWSjzD5Q41jw8LmKwtmjskPH9XW1mrRW",
+				},
+			},
+			// TODO: axiom, gmgn, etc.
+		})
+	if err != nil {
+		panic(err)
+	}
+	defer cleanup()
+
+	txCh := make(chan types.TxSignalBatch, 1000)
+
+	exitSignal := make(chan os.Signal, 1)
+	ctx, cancel := context.WithCancel(context.Background())
+	signal.Notify(exitSignal, syscall.SIGINT, syscall.SIGTERM)
+
+	go func() {
+		<-exitSignal
+		cancel()
+	}()
+
+	// async read from shreder
+	go func() {
+		err := shrederClient.ReadSync(ctx, txCh)
+		if err != nil {
+			if !errors.Is(err, context.Canceled) {
+				panic(err)
+			}
+		}
+	}()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case txBatch := <-txCh:
+			jsonData, _ := json.MarshalIndent(txBatch, "", "  ")
+			fmt.Println(string(jsonData))
+		}
+	}
 }

pkg/logger/logger.go

@@ -16,3 +16,25 @@ type Logger interface {
 	Fatal(a ...interface{})
 	Fatalf(format string, a ...interface{})
 }
+
+type EmptyLogger struct {
+}
+
+func NewEmptyLogger() Logger {
+	return &EmptyLogger{}
+}
+
+func (l *EmptyLogger) Info(a ...interface{})                 {}
+func (l *EmptyLogger) Infof(format string, a ...interface{}) {}
+
+func (l *EmptyLogger) Error(a ...interface{})                 {}
+func (l *EmptyLogger) Errorf(format string, a ...interface{}) {}
+
+func (l *EmptyLogger) Debug(a ...interface{})                 {}
+func (l *EmptyLogger) Debugf(format string, a ...interface{}) {}
+
+func (l *EmptyLogger) Warn(a ...interface{})                 {}
+func (l *EmptyLogger) Warnf(format string, a ...interface{}) {}
+
+func (l *EmptyLogger) Fatal(a ...interface{})                 {}
+func (l *EmptyLogger) Fatalf(format string, a ...interface{}) {}

pkg/shreder/shreder_client.go

@@ -1,4 +1,4 @@
-package shreder_client
+package shreder
 
 import (
 	"context"
@@ -16,9 +16,14 @@ type ShrederClient struct {
 
 	conn         *grpc.ClientConn
 	client       shreder_protos.ShrederServiceClient
+	subscription map[string]*shreder_protos.SubscribeRequestFilterTransactions
 }
 
-func NewShrederClient(logger logger.Logger, url string) (*ShrederClient, func(), error) {
+func NewShrederClient(
+	logger logger.Logger,
+	url string,
+	subscription map[string]*shreder_protos.SubscribeRequestFilterTransactions,
+) (*ShrederClient, func(), error) {
 	conn, err := grpc.NewClient(url, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
 		return nil, func() {}, err
@@ -28,6 +33,7 @@ func NewShrederClient(logger logger.Logger, url string) (*ShrederClient, func(),
 		log:          logger,
 		conn:         conn,
 		client:       shreder_protos.NewShrederServiceClient(conn),
+		subscription: subscription,
 	}
 
 	return s, func() {
@@ -53,28 +59,7 @@ func (c *ShrederClient) ReadSync(ctx context.Context, txCh chan<- types.TxSignal
 	}
 
 	err = stream.Send(&shreder_protos.SubscribeTransactionsRequest{
-		Transactions: map[string]*shreder_protos.SubscribeRequestFilterTransactions{
+		Transactions: c.subscription,
-			"pumpfun": {
-				AccountRequired: []string{
-					"6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P",
-				},
-			},
-			"axiom": {
-				AccountRequired: []string{
-					"F5tfvbLog9VdGUPqBDTT8rgXvTTcq7e5UiGnupL1zvBq",
-				},
-			},
-			"photon": {
-				AccountRequired: []string{
-					"BSfD6SHZigAfDWSjzD5Q41jw8LmKwtmjskPH9XW1mrRW",
-				},
-			},
-			"fjsz": {
-				AccountRequired: []string{
-					"FJsZbftBqRLfF7uqUKpm4s2goDr6xsQ5Q3mN7AFJB6hK",
-				},
-			},
-		},
 	})
 	if err != nil {
 		return err

pkg/types/tx_signal.go

@@ -8,25 +8,26 @@ import (
 )
 
 type TxSignal struct {
-	Source        string
+	Source        string          `json:"source"`
-	TxHash        string
+	TxHash        string          `json:"tx_hash"`
-	Maker         string
+	Maker         string          `json:"maker"`
-	Token0Address string
+	Token0Address string          `json:"token0_address"`
-	Token1Address string
+	Token1Address string          `json:"token1_address"`
-	Token0Amount  decimal.Decimal
+	Token0Amount  decimal.Decimal `json:"token0_amount"`
-	Token1Amount  decimal.Decimal
+	Token1Amount  decimal.Decimal `json:"token1_amount"`
-	Block         uint64
+	Block         uint64          `json:"block"`
-	BlockAt       time.Time
+	BlockAt       time.Time       `json:"block_at"`
-	BlockIndex    uint64
+	BlockIndex    uint64          `json:"block_index"`
-	Event         string
+	Event         string          `json:"event"`
-	Program       string
+	Program       string          `json:"program"`
-	IsProcessed   bool
+	IsProcessed   bool            `json:"is_processed"`
-	IsToken2022   bool
+	IsToken2022   bool            `json:"is_token2022"`
-	IsMayhemMode  bool
+	IsMayhemMode  bool            `json:"is_mayhem_mode"`
-	TxFee         decimal.Decimal
+	TxFee         decimal.Decimal `json:"tx_fee"`
 
-	Token0AmountUint64 uint64
+	// parsed values
-	Token1AmountUint64 uint64
+	Token0AmountUint64 uint64 `json:"-"`
+	Token1AmountUint64 uint64 `json:"-"`
 }
 
 func (t *TxSignal) Parse() *TxSignal {