diff options
| author | Baitinq <manuelpalenzuelamerino@gmail.com> | 2024-05-04 21:22:54 +0200 |
|---|---|---|
| committer | Baitinq <manuelpalenzuelamerino@gmail.com> | 2024-05-04 21:22:54 +0200 |
| commit | 642195d99e0e8fa862ec476573a7064d91ffce36 (patch) | |
| tree | b178f045fbf6730161b9f4335604df1825f14c23 /src/payload-processor/processor/processor.go | |
| parent | payload-processor: add concurrency (diff) | |
| download | fs-tracer-backend-642195d99e0e8fa862ec476573a7064d91ffce36.tar.gz fs-tracer-backend-642195d99e0e8fa862ec476573a7064d91ffce36.tar.bz2 fs-tracer-backend-642195d99e0e8fa862ec476573a7064d91ffce36.zip | |
payload-processor: set up testing
Diffstat (limited to 'src/payload-processor/processor/processor.go')
| -rw-r--r-- | src/payload-processor/processor/processor.go | 11 |
1 files changed, 10 insertions, 1 deletions
diff --git a/src/payload-processor/processor/processor.go b/src/payload-processor/processor/processor.go index f2e4e80..fc6792a 100644 --- a/src/payload-processor/processor/processor.go +++ b/src/payload-processor/processor/processor.go @@ -63,7 +63,16 @@ func (p Processor) process(ctx context.Context, cancel context.CancelFunc) { cancel() break } - fmt.Printf("(%s): message at offset %d: %s = %s\n", time.Now().String(), m.Offset, string(m.Key), string(m.Value)) + err = p.handleMessage(m) + if err != nil { + log.Println("failed to handle message:", err) + continue + } p.kafka_reader.CommitMessages(ctx, m) } } + +func (p Processor) handleMessage(m kafka.Message) error { + fmt.Printf("(%s): message at offset %d: %s = %s\n", time.Now().String(), m.Offset, string(m.Key), string(m.Value)) + return nil +} |