Skip to content

feat(csharp): add TCP/TLS producer and consumer examples #2821

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
atharvalade wants to merge 2 commits into
base: master
Choose a base branch
from
Open

feat(csharp): add TCP/TLS producer and consumer examples #2821

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
3bb3a69
Add TCP/TLS producer and consumer examples for C# SDK
atharvalade Feb 26, 2026
096ce74
ci: retrigger checks
atharvalade Feb 26, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions examples/csharp/Iggy_SDK.Examples.sln
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,12 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Iggy_SDK.Examples.NewSdk.Co
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Iggy_SDK.Examples.NewSdk.Producer", "src\NewSdk\Iggy_SDK.Examples.NewSdk.Producer\Iggy_SDK.Examples.NewSdk.Producer.csproj", "{DB9EAC21-61C8-449A-95A9-6DC369B15C02}"
EndProject
Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "TcpTls", "TcpTls", "{A1B2C3D4-E5F6-7890-ABCD-EF1234567890}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Iggy_SDK.Examples.TcpTls.Producer", "src\TcpTls\Iggy_SDK.Examples.TcpTls.Producer\Iggy_SDK.Examples.TcpTls.Producer.csproj", "{B2C3D4E5-F6A7-8901-BCDE-F12345678901}"
EndProject
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Iggy_SDK.Examples.TcpTls.Consumer", "src\TcpTls\Iggy_SDK.Examples.TcpTls.Consumer\Iggy_SDK.Examples.TcpTls.Consumer.csproj", "{C3D4E5F6-A7B8-9012-CDEF-123456789012}"
EndProject
Global
GlobalSection(SolutionConfigurationPlatforms) = preSolution
Debug|Any CPU = Debug|Any CPU
Expand Down Expand Up @@ -95,6 +101,14 @@ Global
{DB9EAC21-61C8-449A-95A9-6DC369B15C02}.Debug|Any CPU.Build.0 = Debug|Any CPU
{DB9EAC21-61C8-449A-95A9-6DC369B15C02}.Release|Any CPU.ActiveCfg = Release|Any CPU
{DB9EAC21-61C8-449A-95A9-6DC369B15C02}.Release|Any CPU.Build.0 = Release|Any CPU
{B2C3D4E5-F6A7-8901-BCDE-F12345678901}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{B2C3D4E5-F6A7-8901-BCDE-F12345678901}.Debug|Any CPU.Build.0 = Debug|Any CPU
{B2C3D4E5-F6A7-8901-BCDE-F12345678901}.Release|Any CPU.ActiveCfg = Release|Any CPU
{B2C3D4E5-F6A7-8901-BCDE-F12345678901}.Release|Any CPU.Build.0 = Release|Any CPU
{C3D4E5F6-A7B8-9012-CDEF-123456789012}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
{C3D4E5F6-A7B8-9012-CDEF-123456789012}.Debug|Any CPU.Build.0 = Debug|Any CPU
{C3D4E5F6-A7B8-9012-CDEF-123456789012}.Release|Any CPU.ActiveCfg = Release|Any CPU
{C3D4E5F6-A7B8-9012-CDEF-123456789012}.Release|Any CPU.Build.0 = Release|Any CPU
EndGlobalSection
GlobalSection(NestedProjects) = preSolution
{6CA0C169-109F-4A43-A21F-0792C1F865C8} = {3576C181-8BD5-4845-9CDA-835B33A29A29}
Expand All @@ -113,6 +127,9 @@ Global
{85F31923-45DF-4AEA-A28E-D37CF8E61A78} = {3576C181-8BD5-4845-9CDA-835B33A29A29}
{762203E3-A41B-414A-93BF-B963EAE447D1} = {85F31923-45DF-4AEA-A28E-D37CF8E61A78}
{DB9EAC21-61C8-449A-95A9-6DC369B15C02} = {85F31923-45DF-4AEA-A28E-D37CF8E61A78}
{A1B2C3D4-E5F6-7890-ABCD-EF1234567890} = {3576C181-8BD5-4845-9CDA-835B33A29A29}
{B2C3D4E5-F6A7-8901-BCDE-F12345678901} = {A1B2C3D4-E5F6-7890-ABCD-EF1234567890}
{C3D4E5F6-A7B8-9012-CDEF-123456789012} = {A1B2C3D4-E5F6-7890-ABCD-EF1234567890}
EndGlobalSection
EndGlobal

17 changes: 17 additions & 0 deletions ...arp/src/TcpTls/Iggy_SDK.Examples.TcpTls.Consumer/Iggy_SDK.Examples.TcpTls.Consumer.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net10.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<WarningsAsErrors>true</WarningsAsErrors>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\..\..\foreign\csharp\Iggy_SDK\Iggy_SDK.csproj"/>
<ProjectReference Include="..\..\Iggy_SDK.Examples.Shared\Iggy_SDK.Examples.Shared.csproj"/>
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Logging" Version="10.0.2" />
<PackageReference Include="Microsoft.Extensions.Logging.Console" Version="10.0.2" />
</ItemGroup>
</Project>
113 changes: 113 additions & 0 deletions examples/csharp/src/TcpTls/Iggy_SDK.Examples.TcpTls.Consumer/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,113 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.

// TCP/TLS Consumer Example
//
// Demonstrates consuming messages over a TLS-encrypted TCP connection.
//
// Prerequisites:
// Start the Iggy server with TLS enabled:
// IGGY_TCP_TLS_ENABLED=true
// IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem
// IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem

using System.Text;
using Apache.Iggy;
using Apache.Iggy.Configuration;
using Apache.Iggy.Contracts;
using Apache.Iggy.Enums;
using Apache.Iggy.Factory;
using Apache.Iggy.Kinds;
using Microsoft.Extensions.Logging;

const string streamName = "tls-stream";
const string topicName = "tls-topic";
const uint partitionId = 0;
const uint batchesLimit = 5;

var loggerFactory = LoggerFactory.Create(b => { b.AddConsole(); });
var logger = loggerFactory.CreateLogger<Program>();

// Configure the client with TLS.
// TlsSettings.Enabled = true activates TLS on the TCP transport
// TlsSettings.Hostname = the expected server hostname for certificate verification
// TlsSettings.CertificatePath = path to CA or server certificate PEM file
var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
BaseAddress = "127.0.0.1:8090",
Protocol = Protocol.Tcp,
LoggerFactory = loggerFactory,
TlsSettings = new TlsSettings
{
Enabled = true,
Hostname = "localhost",
CertificatePath = "core/certs/iggy_ca_cert.pem"
}
});

await client.ConnectAsync();
await client.LoginUser("iggy", "iggy");
logger.LogInformation("Connected and logged in over TLS.");

await ConsumeMessages();

async Task ConsumeMessages()
{
var interval = TimeSpan.FromMilliseconds(500);
logger.LogInformation(
"Messages will be consumed from stream: {StreamName}, topic: {TopicName}, partition: {PartitionId} with interval {Interval}.",
streamName, topicName, partitionId, interval);

var offset = 0ul;
uint messagesPerBatch = 10;
var consumedBatches = 0;
var consumer = Consumer.New(1);
while (true)
{
if (consumedBatches == batchesLimit)
{
logger.LogInformation("Consumed {ConsumedBatches} batches of messages, exiting.", consumedBatches);
return;
}

var polledMessages = await client.PollMessagesAsync(
Identifier.String(streamName),
Identifier.String(topicName),
partitionId,
consumer,
PollingStrategy.Offset(offset),
messagesPerBatch,
false);

if (!polledMessages.Messages.Any())
{
logger.LogInformation("No messages found.");
await Task.Delay(interval);
continue;
}

offset += (ulong)polledMessages.Messages.Count;
foreach (var message in polledMessages.Messages)
{
var payload = Encoding.UTF8.GetString(message.Payload);
logger.LogInformation("Handling message at offset: {Offset}, payload: {Payload}...",
message.Header.Offset, payload);
}
consumedBatches++;
await Task.Delay(interval);
}
}
17 changes: 17 additions & 0 deletions ...arp/src/TcpTls/Iggy_SDK.Examples.TcpTls.Producer/Iggy_SDK.Examples.TcpTls.Producer.csproj
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
<Project Sdk="Microsoft.NET.Sdk">
<PropertyGroup>
<OutputType>Exe</OutputType>
<TargetFramework>net10.0</TargetFramework>
<ImplicitUsings>enable</ImplicitUsings>
<Nullable>enable</Nullable>
<WarningsAsErrors>true</WarningsAsErrors>
</PropertyGroup>
<ItemGroup>
<ProjectReference Include="..\..\..\..\..\foreign\csharp\Iggy_SDK\Iggy_SDK.csproj"/>
<ProjectReference Include="..\..\Iggy_SDK.Examples.Shared\Iggy_SDK.Examples.Shared.csproj"/>
</ItemGroup>
<ItemGroup>
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.2" />
<PackageReference Include="Microsoft.Extensions.Logging.Console" Version="10.0.2" />
</ItemGroup>
</Project>
141 changes: 141 additions & 0 deletions examples/csharp/src/TcpTls/Iggy_SDK.Examples.TcpTls.Producer/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.

// TCP/TLS Producer Example
//
// Demonstrates producing messages over a TLS-encrypted TCP connection.
//
// Prerequisites:
// Start the Iggy server with TLS enabled:
// IGGY_TCP_TLS_ENABLED=true
// IGGY_TCP_TLS_CERT_FILE=core/certs/iggy_cert.pem
// IGGY_TCP_TLS_KEY_FILE=core/certs/iggy_key.pem

using System.Text;
using Apache.Iggy;
using Apache.Iggy.Configuration;
using Apache.Iggy.Enums;
using Apache.Iggy.Exceptions;
using Apache.Iggy.Factory;
using Apache.Iggy.Messages;
using Microsoft.Extensions.Logging;
using Partitioning = Apache.Iggy.Kinds.Partitioning;

const string streamName = "tls-stream";
const string topicName = "tls-topic";
const uint partitionId = 0;
const uint batchesLimit = 5;

var loggerFactory = LoggerFactory.Create(b => { b.AddConsole(); });
var logger = loggerFactory.CreateLogger<Program>();

// Configure the client with TLS.
// TlsSettings.Enabled = true activates TLS on the TCP transport
// TlsSettings.Hostname = the expected server hostname for certificate verification
// TlsSettings.CertificatePath = path to CA or server certificate PEM file
var client = IggyClientFactory.CreateClient(new IggyClientConfigurator
{
BaseAddress = "127.0.0.1:8090",
Protocol = Protocol.Tcp,
LoggerFactory = loggerFactory,
TlsSettings = new TlsSettings
{
Enabled = true,
Hostname = "localhost",
CertificatePath = "core/certs/iggy_ca_cert.pem"
}
});

await client.ConnectAsync();
await client.LoginUser("iggy", "iggy");
logger.LogInformation("Connected and logged in over TLS.");

await InitSystem();
await ProduceMessages();

async Task InitSystem()
{
try
{
await client.CreateStreamAsync(streamName);
logger.LogInformation("Stream was created.");
}
catch (Exception ex) when (ex is InvalidResponseException or IggyInvalidStatusCodeException)
{
logger.LogWarning("Stream already exists and will not be created again.");
}

try
{
await client.CreateTopicAsync(
Identifier.String(streamName),
topicName,
1,
CompressionAlgorithm.None
);
logger.LogInformation("Topic was created.");
}
catch (Exception ex) when (ex is InvalidResponseException or IggyInvalidStatusCodeException)
{
logger.LogWarning("Topic already exists and will not be created again.");
}
}

async Task ProduceMessages()
{
var interval = TimeSpan.FromMilliseconds(500);
logger.LogInformation(
"Messages will be sent to stream: {StreamName}, topic: {TopicName}, partition: {PartitionId} with interval {Interval}.",
streamName, topicName, partitionId, interval);

var currentId = 0;
var messagesPerBatch = 10;
var sentBatches = 0;
var partitioning = Partitioning.PartitionId((int)partitionId);
while (true)
{
if (sentBatches == batchesLimit)
{
logger.LogInformation("Sent {SentBatches} batches of messages, exiting.", sentBatches);
return;
}

var payloads = Enumerable
.Range(currentId, messagesPerBatch)
.Aggregate(new List<string>(), (list, next) =>
{
list.Add($"message-{next}");
return list;
});

var messages = payloads
.Select(payload => new Message(Guid.NewGuid(), Encoding.UTF8.GetBytes(payload)))
.ToList();

await client.SendMessagesAsync(
Identifier.String(streamName),
Identifier.String(topicName),
partitioning,
messages);

currentId += messagesPerBatch;
sentBatches++;
logger.LogInformation("Sent messages: {Messages}.", payloads);

await Task.Delay(interval);
}
}