Implement RockDB Storage #68
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| using MinimalKafka.Builders; | ||
| using MinimalKafka.Extension; | ||
|
|
||
| namespace KafkaAdventure.Extensions; | ||
|
|
||
| public static class KafkaBuilderExtensions | ||
| { | ||
| public static IKafkaConventionBuilder AsFeature(this IKafkaConventionBuilder builder, string featureName) | ||
| { | ||
| builder.WithClientId(featureName); | ||
| builder.WithGroupId(featureName + "-" + Environment.GetEnvironmentVariable("ASPNETCORE_ENVIRONMENT")); | ||
| return builder; | ||
| } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| using KafkaAdventure.Extensions; | ||
| using MinimalKafka.Extension; | ||
| using MinimalKafka.Stream; | ||
|
|
||
| namespace KafkaAdventure.Features.CommandProcessor; | ||
|
|
||
| public static class ProcessorFeature | ||
| { | ||
| public static void MapProcessor(this WebApplication app) | ||
| { | ||
| Console.WriteLine("Starting Up ProcessorFeature"); | ||
|
|
||
| app.MapStream<string, Command>("game-commands") | ||
| .SplitInto(x => | ||
| { | ||
| x.Branch( | ||
| (k, v) => v.IsCommand("HELP"), | ||
| (c,k,v) => c.ProduceAsync("game-response", k, | ||
| new Response(v.Cmd, "Commands: go [north/south/east/west], look, take [item], inventory")) | ||
| ); | ||
|
|
||
| x.Branch( | ||
| (k, v) => v.IsCommand("GO"), | ||
| (c, k, v) => c.ProduceAsync("game-movement", k, | ||
| new { v.Cmd, Direction = string.Join("", v.Args) }) | ||
| ); | ||
|
Comment on lines
+24
to
+26
Contributor
There was a problem hiding this comment. 🛠️ Refactor suggestion Review the direction concatenation logic. The Consider using a space separator or taking only the first argument: -new { v.Cmd, Direction = string.Join("", v.Args) })
+new { v.Cmd, Direction = string.Join(" ", v.Args) })Or if only single-word directions are expected: -new { v.Cmd, Direction = string.Join("", v.Args) })
+new { v.Cmd, Direction = v.Args.FirstOrDefault() ?? "" })🤖 Prompt for AI Agents |
||
|
|
||
| x.Branch( | ||
| (k, v) => v.IsCommand("LOOK"), | ||
| (c, k, v) => c.ProduceAsync("game-movement", k, | ||
| new { Cmd = "GO", Direction = "LOOK" } | ||
| ) | ||
| ); | ||
|
|
||
| x.Branch( | ||
| (k, v) => v.IsCommand("INVENTORY"), | ||
| (c, k, v) => c.ProduceAsync("game-inventory", k, v) | ||
| ); | ||
|
|
||
| x.Branch( | ||
| (k, v) => v.IsCommand("ECHO"), | ||
| (c, k, v) => c.ProduceAsync("game-response", k, | ||
| new Response(v.Cmd, $"ECHO: {string.Join(' ', v.Args)}")) | ||
| ); | ||
|
|
||
| x.DefaultBranch((c, k, v) | ||
| => c.ProduceAsync("game-response", k, | ||
| new Response(v.Cmd, $"The command '{v.Cmd}' is invalid!"))); | ||
| }) | ||
| .AsFeature("Commands"); | ||
| } | ||
|
|
||
| public record Command(string Cmd, string[] Args) | ||
| { | ||
| public bool IsCommand(string s) => Cmd.StartsWith(s, StringComparison.InvariantCultureIgnoreCase); | ||
| }; | ||
|
|
||
| public record Response(string Command, string Value); | ||
| } | ||
| Original file line number | Diff line number | Diff line change | ||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| @@ -0,0 +1,26 @@ | ||||||||||||||||||||||||||||
| using KafkaAdventure.Extensions; | ||||||||||||||||||||||||||||
| using Microsoft.AspNetCore.Mvc; | ||||||||||||||||||||||||||||
| using Microsoft.AspNetCore.SignalR; | ||||||||||||||||||||||||||||
| using MinimalKafka; | ||||||||||||||||||||||||||||
| using MinimalKafka.Metadata; | ||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||
| namespace KafkaAdventure.Features.Input; | ||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||
| public static class InputFeature | ||||||||||||||||||||||||||||
| { | ||||||||||||||||||||||||||||
| public static void MapInput<T>(this T app) | ||||||||||||||||||||||||||||
| where T : IEndpointRouteBuilder, IApplicationBuilder | ||||||||||||||||||||||||||||
| { | ||||||||||||||||||||||||||||
| Console.WriteLine("Starting Up InputFeature"); | ||||||||||||||||||||||||||||
| app.MapHub<InputHub>("/input"); | ||||||||||||||||||||||||||||
| app.MapTopic("game-response", HandleAsync) | ||||||||||||||||||||||||||||
| .AsFeature("Input"); | ||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||
| public static async Task HandleAsync([FromServices] IHubContext<InputHub> hub, [FromKey] string gameId, [FromValue] Response response) | ||||||||||||||||||||||||||||
| { | ||||||||||||||||||||||||||||
| await hub.Clients.Group(gameId).SendAsync("ReceiveMessage", response.Value); | ||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||
|
Comment on lines
+20
to
+23
Contributor
There was a problem hiding this comment. 🛠️ Refactor suggestion Add error handling to the HandleAsync method The public static async Task HandleAsync([FromServices] IHubContext<InputHub> hub, [FromKey] string gameId, [FromValue] Response response)
{
+ try {
await hub.Clients.Group(gameId).SendAsync("ReceiveMessage", response.Value);
+ } catch (Exception ex) {
+ // Get ILogger from DI or use a static logger
+ Console.Error.WriteLine($"Error sending message to group {gameId}: {ex.Message}");
+ }
}📝 Committable suggestion
Suggested change
|
||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||
| public record Response(string Command, string Value); | ||||||||||||||||||||||||||||
|
Contributor
There was a problem hiding this comment. Duplicate record definition - Response is already defined in InputHub.cs The - public record Response(string Command, string Value);📝 Committable suggestion
Suggested change
|
||||||||||||||||||||||||||||
| } | ||||||||||||||||||||||||||||
| Original file line number | Diff line number | Diff line change | ||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| @@ -0,0 +1,36 @@ | ||||||||||||||||
| using Confluent.Kafka; | ||||||||||||||||
| using Microsoft.AspNetCore.SignalR; | ||||||||||||||||
|
|
||||||||||||||||
| namespace KafkaAdventure.Features; | ||||||||||||||||
|
|
||||||||||||||||
| public class InputHub( | ||||||||||||||||
| IProducer<string, Response> response, | ||||||||||||||||
|
Check warning on line 7 in examples/KafkaAdventure/Features/Input/InputHub.cs
|
||||||||||||||||
| IProducer<string, Command> command | ||||||||||||||||
| ) : Hub | ||||||||||||||||
|
Comment on lines
+6
to
+9
Contributor
There was a problem hiding this comment. Unused parameter warning - the 'response' parameter is never used The constructor parameter public class InputHub(
- IProducer<string, Response> response,
IProducer<string, Command> command
) : Hub📝 Committable suggestion
Suggested change
🧰 Tools🪛 GitHub Check: Build[warning] 7-7: [warning] 7-7: |
||||||||||||||||
| { | ||||||||||||||||
| public async Task JoinGame(string gameId) | ||||||||||||||||
| { | ||||||||||||||||
| await Groups.AddToGroupAsync(Context.ConnectionId, gameId); | ||||||||||||||||
|
|
||||||||||||||||
| await Clients.Group(gameId).SendAsync("ReceiveMessage", "Welcome to 'a Kafka Adventure'"); | ||||||||||||||||
| await Clients.Group(gameId).SendAsync("ReceiveMessage", "You are an aspiring adventurer in search of the legendary relic."); | ||||||||||||||||
| await Clients.Group(gameId).SendAsync("ReceiveMessage", "Type your commands to explore the world. Type 'help' for a list of commands."); | ||||||||||||||||
| } | ||||||||||||||||
|
|
||||||||||||||||
| public async Task SendMessage(string gameId, string message) | ||||||||||||||||
| { | ||||||||||||||||
| if(string.IsNullOrWhiteSpace(message)) | ||||||||||||||||
| { | ||||||||||||||||
| return; | ||||||||||||||||
| } | ||||||||||||||||
|
|
||||||||||||||||
| var cmd = message.Split(' '); | ||||||||||||||||
| await command.ProduceAsync("game-commands", new() | ||||||||||||||||
| { | ||||||||||||||||
| Key = gameId, | ||||||||||||||||
| Value = new Command(cmd.First(), [.. cmd.Skip(1)]) | ||||||||||||||||
| }); | ||||||||||||||||
| } | ||||||||||||||||
| } | ||||||||||||||||
| public record Response(string Command, string Value); | ||||||||||||||||
| public record Command(string cmd, string[] Args); | ||||||||||||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,9 @@ | ||
| namespace KafkaAdventure.Features.Locations; | ||
|
|
||
| public class Location | ||
| { | ||
| public int Id { get; set; } | ||
| public required string Name { get; set; } | ||
| public required string Description { get; set; } | ||
| public required Dictionary<string, string> Exits { get; set; } | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,28 @@ | ||
| using System.Reflection; | ||
| using System.Text.Json; | ||
|
|
||
| namespace KafkaAdventure.Features.Locations; | ||
|
|
||
| public class LocationContext | ||
| { | ||
| public LocationContext() | ||
| { | ||
| Locations = [ | ||
| new Location() { | ||
| Id = 1, | ||
| Name = "The Forest", | ||
| Description = "You are in a dark forest. The trees are tall and the air is damp.", | ||
| Exits = new Dictionary<string, string> { | ||
| { "north", "The Forest" }, | ||
| { "south", "The Forest" }, | ||
| { "east", "The Forest" }, | ||
| { "west", "The Forest" } | ||
| } | ||
| }]; | ||
|
|
||
| } | ||
|
|
||
| public List<Location> Locations { get; set; } | ||
|
|
||
|
|
||
| } |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🛠️ Refactor suggestion
Add error handling for async operations.
The
ProduceAsynccalls lack error handling, which could lead to silent failures in production environments.Consider adding try-catch blocks or using a consistent error handling pattern:
Also applies to: 24-25, 30-32, 37-37, 42-43, 47-48
🤖 Prompt for AI Agents