add on error

Author: LiorBanai

Analogy.Implementation.KafkaProvider.Example/AnalogyKafkaExampleDataProvider.cs

@@ -34,7 +34,11 @@ public void StartReceiving()
 
         public void StopReceiving()
         {
+            sim.Stop();
             Consumer.StopConsuming();
+            Consumer.OnMessageReady -= Consumer_OnMessageReady;
+            Consumer.OnError -= Consumer_OnError;
+
         }
 
         public void InitDataProvider()
@@ -43,11 +47,18 @@ public void InitDataProvider()
             Producer = new KafkaProducer<AnalogyLogMessage>(kafkaUrl, topic, new KafkaSerializer<AnalogyLogMessage>());
             Consumer = new KafkaConsumer<AnalogyLogMessage>(groupId, kafkaUrl, topic);
             Consumer.OnMessageReady += Consumer_OnMessageReady;
+            Consumer.OnError += Consumer_OnError;
             sim = new TimerMessagesSimulator(async m => { await Producer.PublishAsync(m); });
             IsConnected = true;
 
         }
 
+        private void Consumer_OnError(object sender, KafkaMessageArgs<string> e)
+        {
+            AnalogyLogMessage error = new AnalogyLogMessage(e.Message, AnalogyLogLevel.Error, AnalogyLogClass.General, Environment.MachineName);
+            OnMessageReady?.Invoke(sender, new AnalogyLogMessageArgs(error, Environment.MachineName, Environment.MachineName, ID));
+        }
+
         private void Consumer_OnMessageReady(object sender, KafkaMessageArgs<AnalogyLogMessage> e)
         {
             OnMessageReady?.Invoke(sender, new AnalogyLogMessageArgs(e.Message, Environment.MachineName, Environment.MachineName, ID));

Analogy.Implementation.KafkaProvider.Example/TimerMessagesSimulator.cs

@@ -27,5 +27,6 @@ public TimerMessagesSimulator(Action<AnalogyLogMessage> action)
         }
 
         public void Start() => SimulateOnlineMessages.Start();
+        public void Stop() => SimulateOnlineMessages.Stop();
     }
 }

Analogy.Implementation.KafkaProvider/Analogy.Implementation.KafkaProvider.csproj

@@ -2,7 +2,7 @@
 
   <PropertyGroup>
     <TargetFramework>netstandard2.0</TargetFramework>
-    <Version>0.1.1</Version>
+    <Version>0.1.2</Version>
     <Authors>Lior Banai</Authors>
     <Company>Lior Banai</Company>
     <Product>Analogy.Implementation.KafkaProvider</Product>

Analogy.Implementation.KafkaProvider/Analogy.Implementation.KafkaProvider.nuspec

@@ -2,7 +2,7 @@
 <package xmlns="http://schemas.microsoft.com/packaging/2010/07/nuspec.xsd">
   <metadata>
     <id>Analogy.LogViewer.KafkaProvider</id>
-    <version>0.1.1</version>
+    <version>0.1.2</version>
     <title>Analogy Log Viewer Kafka Provider</title>
     <authors>Lior Banai</authors>
     <owners>Lior Banai</owners>

Analogy.Implementation.KafkaProvider/KafkaConsumer.cs

@@ -14,6 +14,7 @@ public class KafkaConsumer<T>
         private string Topic { get; set; }
         private ConsumerConfig Config { get; set; }
         public event EventHandler<KafkaMessageArgs<T>> OnMessageReady;
+        public event EventHandler<KafkaMessageArgs<string>> OnError;
         public BlockingCollectionQueue<T> Queue;
         public BlockingCollectionQueue<string> ErrorsQueue;
         private readonly KafkaSerializer<T> serializer;
@@ -54,12 +55,19 @@ private Task ConsumeAsync()
                              }
                              catch (TaskCanceledException ce)
                              {
+                                 string error = $"TaskCanceledException occurred. Exception: {ce}";
+                                 ErrorsQueue.Enqueue(error);
+                                 OnError?.Invoke(this, new KafkaMessageArgs<string>(error));
                                  Queue.CompleteAdding();
                                  return;
                              }
                              catch (ConsumeException e)
                              {
-                                 ErrorsQueue.Enqueue($"Error occurred: {e.Error.Reason}");
+                                 string error = $"Error occurred: {e.Error.Reason}. Exception: {e}";
+                                 ErrorsQueue.Enqueue(error);
+                                 OnError?.Invoke(this, new KafkaMessageArgs<string>(error));
+                                 Queue.CompleteAdding();
+                                 return;
                              }
                          }
                      }
@@ -84,6 +92,7 @@ private Task ReadAsync() => Task.Factory.StartNew(() =>
         public void StopConsuming()
         {
             cts.Cancel();
+            Queue.CompleteAdding();
         }