more cleanup

Author: metachris

api/handler.go

@@ -41,6 +41,7 @@ func (s *Server) handleTxSSE(w http.ResponseWriter, r *http.Request) {
 			return
 
 		case tx := <-subscriber.txC:
+			// Note/TODO: a client with a slow connection may cause blocking other clients and cause DoS on all receivers
 			fmt.Fprintf(w, "data: %s\n\n", tx)
 			w.(http.Flusher).Flush() //nolint:forcetypeassert
 

collector/clickhouse.go

@@ -235,21 +235,3 @@ func (ch *Clickhouse) sendBatchWithRetries(name string, batch driver.Batch) {
 		metrics.IncClickhouseBatchSaveRetries()
 	}
 }
-
-// FlishData flushes any remaining transactions and source logs in the batch to Clickhouse.
-// This is useful for ensuring all data is saved before shutting down the collector.
-// Needs a lock to avoid concurrent access to the batches.
-//
-// func (ch *Clickhouse) FlushData() {
-// 	// Flush any remaining transactions in the batch
-// 	if len(ch.currentTxBatch) > 0 {
-// 		ch.saveTxs(slices.Clone(ch.currentTxBatch))
-// 		ch.currentTxBatch = ch.currentTxBatch[:0] // Clear the slice without reallocating
-// 	}
-
-// 	// Flush any remaining source logs in the batch
-// 	if len(ch.currentSourcelogBatch) > 0 {
-// 		ch.saveSourcelogs(slices.Clone(ch.currentSourcelogBatch))
-// 		ch.currentSourcelogBatch = ch.currentSourcelogBatch[:0] // Clear the slice without reallocating
-// 	}
-// }

collector/tx_processor.go

@@ -166,6 +166,9 @@ func (p *TxProcessor) Start() {
 func (p *TxProcessor) startTransactionReceiverLoop() {
 	p.log.Info("Waiting for transactions...")
 	for txIn := range p.txC {
+		if txIn.Tx == nil {
+			continue
+		}
 		p.processTx(txIn)
 	}
 }