Refactor to make consumer stages to use base classes

src/Akka.Streams.Kafka.Tests/KafkaFixture.cs

@@ -72,6 +72,15 @@ public async Task InitializeAsync()
                 ["KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR"] = "1"
             });
 
+            /*
+            // Remove old networks, if there are any
+            var networks = await _client.Networks.ListNetworksAsync(new NetworksListParameters());
+            foreach (var existingNetwork in networks.Where(n => n.Name.StartsWith("network-")))
+            {
+                await _client.Networks.DeleteNetworkAsync(existingNetwork.ID);
+            }
+            */
+
             // Setting up network for containers to communicate
             var network = await _client.Networks.CreateNetworkAsync(new NetworksCreateParameters(new NetworkCreate())
             {

src/Akka.Streams.Kafka/Dsl/KafkaConsumer.cs

@@ -4,6 +4,7 @@
 using Akka.Streams.Kafka.Stages;
 using Confluent.Kafka;
 using Akka.Streams.Kafka.Messages;
+using Akka.Streams.Kafka.Stages.Consumers;
 
 namespace Akka.Streams.Kafka.Dsl
 {
@@ -24,7 +25,7 @@ public static class KafkaConsumer
         /// </summary>
         public static Source<ConsumeResult<K, V>, Task> PlainSource<K, V>(ConsumerSettings<K, V> settings, ISubscription subscription)
         {
-            return Source.FromGraph(new KafkaSourceStage<K, V>(settings, subscription));
+            return Source.FromGraph(new PlainSourceStage<K, V>(settings, subscription));
         }
 
         /// <summary>
@@ -37,7 +38,7 @@ public static Source<ConsumeResult<K, V>, Task> PlainSource<K, V>(ConsumerSettin
         /// </summary>
         public static Source<CommittableMessage<K, V>, Task> CommittableSource<K, V>(ConsumerSettings<K, V> settings, ISubscription subscription)
         {
-            return Source.FromGraph(new CommittableConsumerStage<K, V>(settings, subscription));
+            return Source.FromGraph(new CommittableSourceStage<K, V>(settings, subscription));
         }
     }
 }

src/Akka.Streams.Kafka/Messages/CommittableMessage.cs

@@ -1,6 +1,8 @@
 using System;
 using System.Collections.Generic;
+using System.Threading.Tasks;
 using Akka.Streams.Kafka.Dsl;
+using Akka.Streams.Kafka.Stages.Consumers;
 using Confluent.Kafka;
 
 namespace Akka.Streams.Kafka.Messages
@@ -18,10 +20,19 @@ public CommittableMessage(ConsumeResult<K, V> record, CommitableOffset commitabl
         }
 
         public ConsumeResult<K, V> Record { get; }
-
         public CommitableOffset CommitableOffset { get; }
     }
 
+    /// <summary>
+    /// Commit an offset that is included in a <see cref="CommittableMessage{K,V}"/>
+    /// If you need to store offsets in anything other than Kafka, this API
+    /// should not be used.
+    /// </summary>
+    public interface ICommittable
+    {
+        Task Commit();
+    }
+
     /// <summary>
     /// Included in <see cref="CommittableMessage{K,V}"/>. Makes it possible to
     /// commit an offset or aggregate several offsets before committing.
@@ -30,21 +41,33 @@ public CommittableMessage(ConsumeResult<K, V> record, CommitableOffset commitabl
     /// should be the next message your application will consume,
     /// i.e. lastProcessedMessageOffset + 1.
     /// </summary>
-    public class CommitableOffset
+    public interface ICommittableOffset : ICommittable
     {
-        private readonly Func<List<TopicPartitionOffset>> _task;
+        PartitionOffset Offset { get; }
+    }
 
-        public CommitableOffset(Func<List<TopicPartitionOffset>> task, PartitionOffset offset)
+    public interface ICommittableOffsetMetadata : ICommittableOffset
+    {
+        string Metadata { get; }
+    }
+
+
+    public class CommitableOffset : ICommittableOffsetMetadata
+    {
+        private readonly IInternalCommitter _committer;
+        public PartitionOffset Offset { get; }
+        public string Metadata { get; }
+
+        public CommitableOffset(IInternalCommitter committer, PartitionOffset offset, string metadata)
         {
-            _task = task;
+            _committer = committer;
             Offset = offset;
+            Metadata = metadata;
         }
 
-        public PartitionOffset Offset { get; }
-
-        public List<TopicPartitionOffset> Commit()
+        public Task Commit()
         {
-            return _task();
+            return Task.FromResult(_committer.Commit());
         }
     }
 

src/Akka.Streams.Kafka/Settings/ConsumerSettings.cs

@@ -67,6 +67,8 @@ public ConsumerSettings<TKey, TValue> WithProperty(string key, string value) =>
         public ConsumerSettings<TKey, TValue> WithPollTimeout(TimeSpan pollTimeout) => Copy(pollTimeout: pollTimeout);
 
         public ConsumerSettings<TKey, TValue> WithDispatcher(string dispatcherId) => Copy(dispatcherId: dispatcherId);
+
+        public string GroupId => Properties.ContainsKey("group.id") ? Properties["group.id"] : null;
 
         private ConsumerSettings<TKey, TValue> Copy(
             IDeserializer<TKey> keyDeserializer = null,

src/Akka.Streams.Kafka/Stages/Consumers/Abstract/KafkaSourceStage.cs

@@ -0,0 +1,29 @@
+using System.Threading.Tasks;
+using Akka.Streams.Stage;
+
+namespace Akka.Streams.Kafka.Stages.Consumers
+{
+    public abstract class KafkaSourceStage<K, V, TMessage> : GraphStageWithMaterializedValue<SourceShape<TMessage>, Task>
+    {
+        public string StageName { get; }
+        public Outlet<TMessage> Out { get; } = new Outlet<TMessage>("out");
+        public override SourceShape<TMessage> Shape { get; }
+
+        protected KafkaSourceStage(string stageName)
+        {
+            StageName = stageName;
+            Shape = new SourceShape<TMessage>(Out);
+        }
+
+        protected override Attributes InitialAttributes => Attributes.CreateName(StageName);
+
+        protected abstract GraphStageLogic Logic(SourceShape<TMessage> shape, TaskCompletionSource<NotUsed> completion, Attributes inheritedAttributes);
+
+        public override ILogicAndMaterializedValue<Task> CreateLogicAndMaterializedValue(Attributes inheritedAttributes)
+        {
+            var completion = new TaskCompletionSource<NotUsed>();
+            var result = Logic(Shape, completion, inheritedAttributes);
+            return new LogicAndMaterializedValue<Task>(result, completion.Task);
+        }
+    }
+}