feat: subscriber SubscribeToNewTasksV2

Author: uri-99

aggregator/internal/pkg/aggregator.go

@@ -404,7 +404,6 @@ func (agg *Aggregator) AddNewTask(batchMerkleRoot [32]byte, taskCreatedBlock uin
 	agg.logger.Info("New task added", "batchIndex", batchIndex, "batchIdentifierHash (actually batchMerkleRoot)", "0x"+hex.EncodeToString(batchMerkleRoot[:]))
 }
 func (agg *Aggregator) AddNewTaskV2(batchMerkleRoot [32]byte, senderAddress [20]byte, taskCreatedBlock uint32) {
-
 	batchIdentifier := append(batchMerkleRoot[:], senderAddress[:]...)
 	var batchIdentifierHash = *(*[32]byte)(crypto.Keccak256(batchIdentifier))
 

aggregator/internal/pkg/subscriber.go

@@ -16,10 +16,29 @@ func (agg *Aggregator) SubscribeToNewTasks() error {
 				return err
 			}
 		case newBatch := <-agg.NewBatchChan:
-			agg.AddNewTask(newBatch.BatchMerkleRoot, newBatch.SenderAddress, newBatch.TaskCreatedBlock)
+			agg.AddNewTask(newBatch.BatchMerkleRoot, newBatch.TaskCreatedBlock)
 		}
 	}
+}
+func (agg *Aggregator) SubscribeToNewTasksV2() error {
+	err := agg.subscribeToNewTasks()
+	if err != nil {
+		return err
+	}
+
+	for {
+		select {
+		case err := <-agg.taskSubscriber:
+			agg.AggregatorConfig.BaseConfig.Logger.Info("Failed to subscribe to new tasks", "err", err)
 
+			err = agg.subscribeToNewTasks()
+			if err != nil {
+				return err
+			}
+		case newBatch := <-agg.NewBatchChanV2:
+			agg.AddNewTaskV2(newBatch.BatchMerkleRoot, newBatch.SenderAddress, newBatch.TaskCreatedBlock)
+		}
+	}
 }
 
 func (agg *Aggregator) subscribeToNewTasks() error {

operator/pkg/operator.go

@@ -95,7 +95,7 @@ func NewOperatorFromConfig(configuration config.OperatorConfig) (*Operator, erro
 
 	rpcClient, err := NewAggregatorRpcClient(configuration.Operator.AggregatorServerIpPortAddress, logger)
 	if err != nil {
-		return nil, fmt.Errorf("Could not create RPC client: %s. Is aggregator running?", err)
+		return nil, fmt.Errorf("could not create RPC client: %s. Is aggregator running?", err)
 	}
 
 	operatorId := eigentypes.OperatorIdFromKeyPair(configuration.BlsConfig.KeyPair)