Release

.vscode/settings.json

@@ -22,31 +22,6 @@
  "**/node_modules": true,
  "**/build": true
  },
- "[typescript]": {
- "editor.defaultFormatter": "esbenp.prettier-vscode",
- "editor.codeActionsOnSave": {
- "source.organizeImports": true,
- "source.fixAll": true
- }
- },
- "[typescriptreact]": {
- "editor.defaultFormatter": "esbenp.prettier-vscode",
- "editor.codeActionsOnSave": {
- "source.organizeImports": true,
- "source.fixAll": true
- }
- },
- "typescript.updateImportsOnFileMove.enabled": "always",
- "eslint.enable": true,
- "eslint.lintTask.enable": true,
- "eslint.workingDirectories": [
- {
- "mode": "auto"
- }
- ],
- "files.associations": {
- "*.json": "jsonc"
- },
  "files.exclude": {
  "**/.git": true,
  "**/.hg": true,
@@ -58,10 +33,4 @@
  "editor.tabSize": 2,
  "editor.insertSpaces": true,
  "editor.detectIndentation": false,
- "[typescript][typescriptreact]": {
- "editor.codeActionsOnSave": {
- "source.organizeImports": "explicit",
- "source.fixAll": "explicit"
- }
- },
 }

README.md

@@ -26,12 +26,14 @@ var options = new ProducerBuildOptions
 };
 ```
 
-Then create a new instance of kafka producer use `BuildWithSuperstream()`:
+Then create a new instance of kafka producer and use `SuperstreamInitializer.Init` to initialize the producer with `Superstream` options:
 
 ```c#
-using var producer = new ProducerBuilder<string?, byte[]>(config)
- .BuildWithSuperstream(options);
+var kafkaProducer = new ProducerBuilder<string?, byte[]>(config)
+ .Build();
+using var producer = SuperstreamInitializer.Init(kafkaProducer, options);
 ```
+
 Finally, to produce messages to kafka, use `ProduceAsync` or `Produce`:
 
 ```c#
@@ -57,11 +59,13 @@ var options = new ConsumerBuildOptions
 };
 ```
 
-Then create a new instance of kafka consumer use `BuildWithSuperstream()`:
+Then create a new instance of kafka consumer and use `SuperstreamInitializer.Init` to initialize the consumer with `Superstream` options:
 
 ```c#
-using var consumer = new ConsumerBuilder<Ignore, byte[]>(config)
- .BuildWithSuperstream(options);
+var kafkaConsumer = new ConsumerBuilder<Ignore, byte[]>(config)
+ .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
+ .Build();
+using var consumer = SuperstreamInitializer.Init(kafkaConsumer, options);
 ```
 
 Finally, to consume messages from kafka, use `Consume`:

examples/Consumer/Program.cs

@@ -6,7 +6,7 @@
 var token = "<superstream-token>";
 var host = "<superstream-host>";
 string brokerList = "<brokers>";
-var topics = new List<string> {"test"};
+var topics = new List<string> { "test" };
 
 Console.WriteLine($"Started consumer, Ctrl-C to stop consuming");
 
@@ -16,9 +16,9 @@
  GroupId = "cg",
  BootstrapServers = brokerList,
  EnableAutoCommit = false,
- SaslPassword= "...",
+ SaslPassword = "...",
  SaslUsername = "...",
- SecurityProtocol = SecurityProtocol.SaslSsl, 
+ SecurityProtocol = SecurityProtocol.SaslSsl,
  SaslMechanism = SaslMechanism.Plain
 };
 var options = new ConsumerBuildOptions
@@ -28,9 +28,10 @@
  ConsumerConfig = config
 };
 
-using var consumer = new ConsumerBuilder<Ignore, byte[]>(config)
+var kafkaConsumer = new ConsumerBuilder<Ignore, byte[]>(config)
  .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
- .BuildWithSuperstream(options);
+ .Build();
+using var consumer = SuperstreamInitializer.Init(kafkaConsumer, options);
 
 // consume by specific partition
 consumer.Assign(topics.Select(topic => new TopicPartitionOffset(topic, 1, Offset.Beginning)).ToList());

examples/Producer/Program.cs

@@ -6,15 +6,17 @@
 
 var token = "<superstream-token>";
 var host = "<superstream-host>";
-string brokerList = "<brokers>";
+string brokerList = "";
 string topicName = "topic_1";
 
-var config = new ProducerConfig { 
- BootstrapServers = brokerList, 
- SaslPassword= "...", 
- SaslUsername= "...", 
- SecurityProtocol = SecurityProtocol.SaslSsl, 
- SaslMechanism = SaslMechanism.Plain};
+var config = new ProducerConfig
+{
+ BootstrapServers = brokerList,
+ // SaslPassword= "...", 
+ // SaslUsername= "...", 
+ // SecurityProtocol = SecurityProtocol.SaslSsl, 
+ SaslMechanism = SaslMechanism.Plain
+};
 
 var options = new ProducerBuildOptions
 {
@@ -23,8 +25,11 @@
  ProducerConfig = config,
  LearningFactor = 250 // optional
 };
-using var producer = new ProducerBuilder<string?, byte[]>(config)
- .BuildWithSuperstream(options);
+
+var kafkaProducer = new ProducerBuilder<string?, byte[]>(config)
+ .Build();
+using var producer = SuperstreamInitializer.Init(kafkaProducer, options);
+
 Console.WriteLine("\n-----------------------------------------------------------------------");
 Console.WriteLine($"Producer {producer.Name} producing on topic {topicName}.");
 Console.WriteLine("-----------------------------------------------------------------------");
@@ -49,7 +54,7 @@
  var deliveryReport = await producer.ProduceAsync(
  topicName, new() { Key = key, Value = JsonSerializer.SerializeToUtf8Bytes(person) });
 
- await producer.ProduceAsync("<topic>", new() { Value = Encoding.UTF8.GetBytes("{\"test_key\":\"test_value\"}")});
+ await producer.ProduceAsync("<topic>", new() { Value = Encoding.UTF8.GetBytes("{\"test_key\":\"test_value\"}") });
  Console.WriteLine($"Delivered to : {deliveryReport.TopicPartitionOffset}");
 
  }

src/Superstream/Extensions/SuperstreamInitializerExtensions.cs

@@ -1,8 +1,8 @@
 namespace Superstream;
 
-public static class SuperstreamInitializerExtensions
+internal static class SuperstreamInitializerExtensions
 {
- public static IProducer<TKey, TValue> BuildWithSuperstream<TKey, TValue>(
+ internal static IProducer<TKey, TValue> BuildWithSuperstream<TKey, TValue>(
  this ProducerBuilder<TKey, TValue> builder,
  ProducerBuildOptions options
  )
@@ -17,7 +17,7 @@ ProducerBuildOptions options
  );
  }
 
- public static IConsumer<TKey, TValue> BuildWithSuperstream<TKey, TValue>(
+ internal static IConsumer<TKey, TValue> BuildWithSuperstream<TKey, TValue>(
  this ConsumerBuilder<TKey, TValue> builder,
  ConsumerBuildOptions options
  )

src/Superstream/Interceptors/ConsumerInterceptor.cs

@@ -1,7 +1,8 @@
 namespace Superstream.Interceptors;
 
-internal class ConsumerInterceptor<TKey, TValue> : DispatchProxy
+internal class ConsumerInterceptor<TKey, TValue> : DispatchProxy, IDisposable
 {
+ private bool _disposed = false;
 #nullable disable
  public IConsumer<TKey, TValue> Target { get; set; }
  public SuperstreamClient Client { get; set; }
@@ -20,7 +21,7 @@
  if (typeof(TValue) != typeof(byte[]))
  return result;
  int partition = -1;
  OnConsume(result as ConsumeResult<TKey, byte[]>, partition);
  return result;
  }
 
@@ -133,4 +134,23 @@
  return proxy as IConsumer<K, V>
  ?? throw new InvalidOperationException(typeof(IConsumer<K, V>).Name);
  }
+
+ public void Dispose()
+ {
+ Dispose(true);
+ GC.SuppressFinalize(this);
+ }
+
+ public void Dispose(bool disposing)
+ {
+ if (_disposed)
+ {
+ return;
+ }
+ if (disposing)
+ {
+ Target.Dispose();
+ }
+ _disposed = true;
+ }
 }

src/Superstream/Interceptors/ProducerInterceptor.cs

@@ -1,7 +1,11 @@
 namespace Superstream.Interceptors;
 
-internal class ProducerInterceptor<TKey, TValue> : DispatchProxy
+internal class ProducerInterceptor<TKey, TValue> : DispatchProxy, IDisposable
 {
+ private bool _disposed;
+
+
+
 #nullable disable
  public IProducer<TKey, TValue> Target { get; set; }
  public SuperstreamClient Client { get; set; }
@@ -20,16 +24,16 @@
  if (typeof(TValue) != typeof(byte[]))
  return targetMethod?.Invoke(Target, args);
 
  int partition = default;
  string topic = string.Empty;
  if (args[0].GetType() == typeof(TopicPartition))
  {
  partition = (args[0] as TopicPartition)!.Partition;
  topic = (args[0] as TopicPartition)!.Topic;
  }
  else
  {
  topic = args[0] as string;
  }
  OnSend(topic, args[1] as Message<TKey, byte[]>, partition);
  var result = targetMethod?.Invoke(Target, args);
@@ -96,7 +100,7 @@
  Client.ProducerProtoDescriptor.MasterMessageName
  )
  .GetAwaiter()
  .GetResult() ?? throw new Exception("Failed to serialize message");
  }
  catch (Exception ex)
  {
@@ -148,4 +152,23 @@
  return proxy as IProducer<K, V>
  ?? throw new InvalidOperationException(typeof(IProducer<K, V>).Name);
  }
+
+ public void Dispose()
+ {
+ Dispose(disposing: true);
+ GC.SuppressFinalize(this);
+ }
+
+ protected virtual void Dispose(bool disposing)
+ {
+ if (!_disposed)
+ {
+ if (disposing)
+ {
+ Target.Dispose();
+ }
+
+ _disposed = true;
+ }
+ }
 }

src/Superstream/Superstream.csproj

@@ -7,14 +7,14 @@
  <LangVersion>12</LangVersion>
 
  <PackageId>Superstream</PackageId>
- <Version>1.0.0</Version>
+ <Version>1.0.1</Version>
  <Authors>Memphis.dev team</Authors>
  <Company>Memphis.dev</Company>
  <PackageTags>Superstream</PackageTags>
  <PackageReadmeFile>README.md</PackageReadmeFile>
  <PackageLicenseFile>LICENSE</PackageLicenseFile>
  <EnablePackageValidation>false</EnablePackageValidation>
- <PackageValidationBaselineVersion>1.0.0</PackageValidationBaselineVersion>
+ <PackageValidationBaselineVersion>1.0.1</PackageValidationBaselineVersion>
 
  </PropertyGroup>
 

src/Superstream/SuperstreamInitializer.cs

@@ -1,8 +1,8 @@
 namespace Superstream;
 
-internal static class SuperstreamInitializer
+public static class SuperstreamInitializer
 {
- public static void Init<TKey, TValue>(
+ internal static void Init<TKey, TValue>(
  ref IProducer<TKey, TValue> producer,
  ProducerBuildOptions options
  )
@@ -17,7 +17,7 @@ ProducerBuildOptions options
  );
  }
 
- public static void Init<TKey, TValue>(
+ internal static void Init<TKey, TValue>(
  ref IConsumer<TKey, TValue> consumer,
  ConsumerBuildOptions options
  )
@@ -31,19 +31,48 @@ ConsumerBuildOptions options
  );
  }
 
- public static IConsumer<TKey, TValue> Init<TKey, TValue>(
+ internal static IConsumer<TKey, TValue> Init<TKey, TValue>(
  ConsumerBuilder<TKey, TValue> builder,
  ConsumerBuildOptions options
  )
  {
  return builder.BuildWithSuperstream(options);
  }
 
- public static IProducer<TKey, TValue> Init<TKey, TValue>(
+ internal static IProducer<TKey, TValue> Init<TKey, TValue>(
  ProducerBuilder<TKey, TValue> builder,
  ProducerBuildOptions options
  )
  {
  return builder.BuildWithSuperstream(options);
  }
+
+ public static IProducer<K, V> Init<K, V>(
+ IProducer<K, V> target,
+ ProducerBuildOptions options
+ )
+ {
+ options.EnsureIsValid();
+ return ProducerInterceptor<K, V>.Init(
+ target,
+ options.ProducerConfig,
+ options.Token,
+ options.Host,
+ options.LearningFactor
+ );
+ }
+
+ public static IConsumer<K, V> Init<K, V>(
+ IConsumer<K, V> target,
+ ConsumerBuildOptions options
+ )
+ {
+ return ConsumerInterceptor<K, V>.Init(
+ target,
+ options.ConsumerConfig,
+ options.Token,
+ options.Host,
+ options.LearningFactor
+ );
+ }
 }

src/Superstream/Utils/Constants.cs

@@ -16,6 +16,6 @@ internal static class Subjects
 
 internal static class SdkInfo
 {
- public const string SdkVersion = "1.0.0";
+ public const string SdkVersion = "1.0.1";
  public const string SdkLanguage = "C#";
 }

version.conf

@@ -1 +1 @@
-1.0.0
+1.0.1