CommitWorker processes commit transactions from the jobs channel and sends the results to the results channel. It starts a tracing span, processes each transaction, and records relevant events and errors. Parameters: - ctx context.Context: The context for the operation. - jobs <-chan *model.Transac
(ctx context.Context, jobs <-chan *model.Transaction, results chan<- BatchJobResult, wg *sync.WaitGroup, amount *big.Int)
| 93 | // - wg *sync.WaitGroup: A wait group to synchronize the completion of the worker. |
| 94 | // - amount *big.Int: The amount to be processed in the transaction. |
| 95 | func (l *LedgerForge) CommitWorker(ctx context.Context, jobs <-chan *model.Transaction, results chan<- BatchJobResult, wg *sync.WaitGroup, amount *big.Int) { |
| 96 | ctx, span := tracer.Start(ctx, "CommitWorker") |
| 97 | defer span.End() |
| 98 | |
| 99 | defer wg.Done() |
| 100 | for originalTxn := range jobs { |
| 101 | if !IsInflightTransaction(originalTxn) { |
| 102 | err := fmt.Errorf("transaction is not in inflight status") |
| 103 | results <- BatchJobResult{Error: err} |
| 104 | span.RecordError(err) |
| 105 | continue |
| 106 | } |
| 107 | queuedCommitTxn, err := l.CommitInflightTransaction(ctx, originalTxn.TransactionID, amount) |
| 108 | if err != nil { |
| 109 | results <- BatchJobResult{Error: err} |
| 110 | span.RecordError(err) |
| 111 | continue |
| 112 | } |
| 113 | results <- BatchJobResult{Txn: queuedCommitTxn} |
| 114 | span.AddEvent("Commit processed", trace.WithAttributes(attribute.String("transaction.id", queuedCommitTxn.TransactionID))) |
| 115 | } |
| 116 | } |
| 117 | |
| 118 | // VoidWorker processes void transactions from the jobs channel and sends the results to the results channel. |
| 119 | // It starts a tracing span, processes each transaction, and records relevant events and errors. |