Poor man's Kafka (stream-processing bus)

F# Advent Calendar 2021

This post is part of Sergey Tihon’s Blog F# Advent Calendar in English 2021. Please check out the other posts as well.

2021-12-01-poor-mans-kafka-stream-processing-bus_fsharp-advent-2021.jpg Sergey Tihon's Blog - F# Advent Calendar in English 2021

I have in two other occasions participated in this (fun _ -> async { let! event = Async.AwaitEvent … }). Perhaps if you find this blog post insightful, you might want to have a read of these other two as well:

2016: “Semantic Versioning .NET libraries and NuGet packages”.
2017: “Bloom Filter Sort (bfsort)”.

Background

First of all, lets try to explain the title of this post and why it’s being used. There are two main parts:

Poor man’s: The definition in the Cambridge English Dictionary states that it is: «used to refer to something that is a worse or cheaper version of something else that is mentioned».

Note: It’s not always the case that just because something is cheaper, it’s worse. As an example, lets take the Black Velvet beer cocktails.

2021-12-01-poor-mans-kafka-stream-processing-bus_black-velvet-beer-cocktails.png Champagne or Cider? Pick your poison

Kafka (stream-processing bus): The definition at Wikipedia Apache Kafka states that it: «is a framework implementation of a software bus using stream-processing. It is an open-source software platform developed by the Apache Software Foundation written in Scala and Java. The project aims to provide a unified, high-throughput, low-latency platform for handling real-time data feeds … Kafka uses a binary TCP-based protocol that is optimized for efficiency and relies on a message set abstraction that naturally groups messages together to reduce the overhead of the network roundtrip. This leads to larger network packets, larger sequential disk operations, contiguous memory blocks […] which allows Kafka to turn a bursty stream of random message writes into linear writes».

The key elements are:

Service bus using stream-processing
High-throughput, low-latency platform for handling real-time data feeds
Uses a binary TCP-based protocol that is optimized for efficiency
Groups messages together to reduce the overhead of the network roundtrip
Larger: network packets, sequential disk operations & contiguous memory blocks

which are the features that we will mimic in our Poor man’s implementation.

Note: For a short, concise and comprehensive overview, I would highly suggest you to watch the following video Apache Kafka Explained (Comprehensive Overview) on YouTube, as you will have a better understanding of the terminology I will use later on in this blog post.

2021-12-01-poor-mans-kafka-stream-processing-bus_kafka-explained.png Finematics: Apache Kafka Explained (Comprehensive Overview)

Poor man’s Kafka from scratch

Now that we have an understanding of what we are trying to do, lets implement it from scratch in F# (*).

(*) - There are only a few places where I had to rely on C#: LINQ due to F# Seq.skip that is not safe (throws an Exception if empty) and Console.WriteLine for thread-safety.

Prerequisites

Before we can write any code, we first need to install .NET 6 (core) on my operating system. I have been using for many years NixOS Linux, the best operating system out there without any doubt, and I’m transitioning right now to macOS, due to their new powerful and battery friendly Apple Silicon chips.

For both systems, I’m using nix, which is the package-manager that spawned an operating system. The power of nix, allows you to define shell scripts in a pure and lazy, but sadly untyped, functional programming language. Here is a sample on how you would write a shell script to retrieve a specific version of dotnet:

/* Version from Robert (rycee): */
{ pkgs ? import <nixpkgs> {
  overlays = [(
    self: super: {
      dotnet-sdk-current = super.dotnet-sdk.overrideAttrs (
        old: rec {
          version = "6.0.100";
          src = super.fetchurl {
            url = "https://dotnetcli.azureedge.net/dotnet/Sdk/${version}/dotnet-sdk-${version}-linux-x64.tar.gz";
            sha256 = "8489a798fcd904a32411d64636e2747edf108192e0b65c6c3ccfb0d302da5ecb";
          };
        }
      );
    }
  )];
}}:

with pkgs;

mkShell {
  buildInputs = [
    ncurses # for `clear`
    emacs
    openssl
    dotnet-sdk-current
  ];
  shellHook = ''
    export CLI_DEBUG=1
    export DOTNET_CLI_TELEMETRY_OPTOUT="1"
    export DOTNET_SYSTEM_GLOBALIZATION_INVARIANT=1
    export LD_LIBRARY_PATH=${openssl.out}/lib/
  '';
}

# References:
#
# Download .NET (current):
#
# - https://dotnet.microsoft.com/download/dotnet
#
# - https://dotnet.microsoft.com/download/dotnet/6.0

which can then be stored as shell.nix file and be placed in a folder of a given project. In order to retrieve the specific dotnet version, you just have to execute nix-shell from a terminal and you will now be able use dotnet with the chosen (SDK) version, ONLY from that terminal:

[nix-shell:~/code/dotnet/kafka/kaf]$ dotnet --info
.NET SDK (reflecting any global.json):
 Version:   6.0.100
 Commit:    9e8b04bbff

Runtime Environment:
 OS Name:     nixos
 OS Version:  21.05.4116.46251a79f75
 OS Platform: Linux
 RID:         linux-x64
 Base Path:   /nix/store/40wzcbndr7agwmmhz2pajb7gd0fn07l7-dotnet-sdk-6.0.100/sdk/6.0.100/

Host (useful for support):
  Version: 6.0.0
  Commit:  4822e3c3aa

.NET SDKs installed:
  6.0.100 [/nix/store/40wzcbndr7agwmmhz2pajb7gd0fn07l7-dotnet-sdk-6.0.100/sdk]

.NET runtimes installed:
  Microsoft.AspNetCore.App 6.0.0 [/nix/store/40wzcbndr7agwmmhz2pajb7gd0fn07l7-dotnet-sdk-6.0.100/shared/Microsoft.AspNetCore.App]
  Microsoft.NETCore.App 6.0.0 [/nix/store/40wzcbndr7agwmmhz2pajb7gd0fn07l7-dotnet-sdk-6.0.100/shared/Microsoft.NETCore.App]

To install additional .NET runtimes or SDKs:
  https://aka.ms/dotnet-download

Note: One of the important parts here is the shellHook’s, which will help you disable a few of Microsoft intrusive privacy-misbehavior and enable a few components that usually makes dotnet applications crash in *nix and macOS.

Structure

For the sake of simplicity, I have created a few libraries, suffixed with .fs, which will contain the shared logic between the executables, which are suffixed with .fsx.

[nix-shell:~/code/dotnet/kafka/kaf]$ ls -lh
total 68K
drwxr-xr-x 2 johndoe users 4.0K Nov 28 18:13 client
drwxr-xr-x 2 johndoe users 4.0K Nov 28 23:37 events
drwxr-xr-x 2 johndoe users 4.0K Nov 28 18:00 topics
-rwxr-xr-x 1 johndoe users 7.1K Nov 28 23:35 broker.fsx
-rwxr-xr-x 1 johndoe users 4.1K Nov 28 23:31 clipub.fsx
-rwxr-xr-x 1 johndoe users 5.0K Nov 28 18:16 clisub.fsx
-rw-r--r-- 1 johndoe users  15K Nov 28 23:28 common.fs
-rwxr-xr-x 1 johndoe users 1.8K Nov 28 23:35 create.fsx
-rw-r--r-- 1 johndoe users  801 Nov 28 16:32 domain.fs
-rwxr-xr-x 1 johndoe users  156 Nov 28 18:17 remove.sh
-rw-r--r-- 1 johndoe users  949 Nov 23 05:54 shell.nix

Note: In order to execute F# scripts seemingly across *nix and macOS, the following header will be added on top of the .fsx files as well as making them executable from the terminal with chmod +x ….fsx

#!/usr/bin/env -S dotnet fsi --mlcompatibility --optimize --warnaserror+:25,26

In this section, we will only mention, with some degree of details, what the following files contain. For full view of the implementation, please have a look at the Appendix: Source Code at the end of this blog post.

common: The library mainly contains the TCP Sockets module, the Kafka Protocol and a few helper modules:
- Error module: Wraps dotnet exceptions as F# record types.
- Date module: Provides ISO-8601 UTC timestamps.
- Input module: Exposes logic to wait for a given ConsoleKey to be pressed. Indispensable when working with Async.Choice and terminal CLI.
- Output module: Provides thread-safe writes to the terminal. Sadly, the built-in printf in F#, isn’t.
- UID module: Generates unique identifiers, which are used to ensure that events, aren’t stored in the same location on disk.
- TCP module: Server and client based on Berkeley-sockets. The module is totally agnostic to the Kafka Protocol, therefore it could be re-used for other projects that relies on low-level POSIX-socket communication. The TCP.Server.listen will spawn a new async process for whenever a new socket connection is being established.
- Kafka Protocol module: Contains the data-types and binary-converters from/to as well as the network-package specification. A few tricks are used to ensure that protocol elements are created as expected: Single-case union constructors are made private but then exposed with Active-patterns to allow type deconstruction from consuming clients.
broker: A simple server, without a cluster component, which will interact with both publishing as well as subscribing clients:
- app function: It’s the main (and recursive) application that will be fed to the TCP.Server.listen function. The application will send a welcome message and will expect the client respond by specifying if it’s a publisher or a subscriber as well as the topic. Based on the client response, it will then dispatch it to either the producer or consumer function.
- producer function: While there is a connection with the publisher, the broker will receive (recursively) batches of messages for the given topic. The batches will be stored in their binary form on disk, where filenames will be prefixed with ISO-8601 UTC timestamps, which will allow to serve those batches in a orderly and timely manner.
  
  Serving batches in an ORDERly and timely manner
- consumer function: When a subscriber connects to the broker, it will provide an index, which will limit the amount of batches the broker will initially serve, as it is the subscribers responsibility, to keep a local state of what data they have already received from the broker. Once these batches are sent to the subscriber, the consumer function will then call the created function.
- created function: As with the producer function, while there is a connection with the subscriber, the broker will (recursively) listen to the specified topic folder, so whenever a new batch arrives, it will be sent immediately to the subscriber to provide real-time data feeds.
domain: Very basic event data-types specification, which are not known by the broker, and a random-helper module:
- Domain module: Basic event-type specification.
- Random module: Helper to generate batches of events for testing/demo purposes.
clipub A simple command line interface (CLI) used by publishers to retrieve human-readable events, convert them to a commonly-agreed binary format and send them to the broker:
- json function: Deserializes human-readable events.
- bson function: Serializes events to a binary format.
- app function: It’s the main (and recursive) application that will be fed to the TCP.Client.interact function. The application will wait for the broker to send a welcome message and will then respond by specifying it’s a publisher (0x10uy) as well as the topic. Afterwards it will call the producer function.
- producer function: While there is a connection with the broker, the producer will (recursively) listen to a specified topic folder, so whenever a new batch of events arrive. It will then retrieve them, deserialize the human-readable format and serialize them with the agreed binary format. Once this is done, it will be sent to the broker for storage persistence.
clisub Is another simple command line interface (CLI), that is used by subscribers to retrieve binary formatted batches from the broker, convert them to a human-readable events and store them locally:
- bson function: Deserializes binary formatted batches.
- json function: Serializes to human-readable events.
- app function: It’s the main (and recursive) application that will be fed to the TCP.Client.interact function. The application will wait for the broker to send a welcome message and will then reply by specifying it’s a publisher (0x20uy) as well as the topic, which will contain an index, which is retrieved from a local state to ensure that the broker doesn’t re-send already received batches. Afterwards it will call the consumer function.
- consumer function: While there is a connection with the broker, the consumer will receive (recursively) batches of messages for the given topic. The binary formatted batches will be deserialized and then serialized to a human-readable format and stored locally on disk, in an orderly and timely manner, by prefixing filenames with ISO-8601 UTC timestamps.
  
  Storing locally on disk, in an ORDERly and timely manner
create: A simple helper-tool used to create random batches of human-readable events.
remove: A simple helper-tool to reset server and clients state.

Here we describe the steps, that were used for the demo:

The broker was started:

2021-11-29T21:46:25.2901362Z | STARTED | BROKER | Poor Man's Kafka

Then the publisher was started:

2021-11-29T21:46:30.2332728Z | STARTED | CLIPUB | Poor Man's Kafka
…
2021-11-29T21:46:30.2763232Z | VERBOSE | RECV   | (15,16,[|0uy; …; 114uy|])

Then the subscriber was started:

2021-11-29T21:46:30.9806649Z | STARTED | CLISUB | Poor Man's Kafka
…
2021-11-29T21:46:31.0159547Z | VERBOSE | RECV   | (15,32,[|0uy; …; 114uy|])

Then the event creator (4 and 22) was executed:

2021-11-29T21:46:41.1228100Z | CREATED | EVENTS | Random events: 4
…
2021-11-29T21:46:45.0834605Z | CREATED | EVENTS | Random events: 22

Then the subscriber was stopped:

2021-11-29T21:46:51.1911832Z | STOPPED | CLISUB | Poor Man's Kafka
…
2021-11-29T21:46:57.4568349Z | FAILURE | BROKER | Consumer:

Then the event creator (17 and 16) was executed again:

2021-11-29T21:46:57.4226179Z | CREATED | EVENTS | Random events: 17
…
2021-11-29T21:47:01.4008672Z | CREATED | EVENTS | Random events: 16

Then the subscriber was started again, catching up:

2021-11-29T21:47:26.2142101Z | STARTED | CLISUB | Poor Man's Kafka
2021-11-29T21:47:26.2575274Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000012287)
2021-11-29T21:47:26.3647421Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000012804)
…
2021-11-29T21:47:26.2494125Z | VERBOSE | RECV   | (15,32,[|0uy; …; 114uy|])
2021-11-29T21:47:26.2535665Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000012287)
2021-11-29T21:47:26.2546217Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000012804)

Then the event creator (15 and 9) was executed again:

2021-11-29T21:47:38.5005655Z | CREATED | EVENTS | Random events: 15
…
2021-11-29T21:47:42.4589979Z | CREATED | EVENTS | Random events: 9

broker.fsx

[nix-shell:~/code/dotnet/kafka/kaf]$ ./broker.fsx 
2021-11-29T21:46:25.2901362Z | STARTED | BROKER | Poor Man's Kafka
2021-11-29T21:46:25.2904900Z | VERBOSE | BROKER | Press Enter to exit
2021-11-29T21:46:30.2349689Z | VERBOSE | CLIENT | Connected
2021-11-29T21:46:30.2435542Z | VERBOSE | SEND   | (1,255,[||])
2021-11-29T21:46:30.2763232Z | VERBOSE | RECV   | (15,16,[|0uy; …; 114uy|])
2021-11-29T21:46:30.9806774Z | VERBOSE | CLIENT | Connected
2021-11-29T21:46:30.9808928Z | VERBOSE | SEND   | (1,255,[||])
2021-11-29T21:46:31.0159547Z | VERBOSE | RECV   | (15,32,[|0uy; …; 114uy|])
2021-11-29T21:46:41.2572331Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000003486)
2021-11-29T21:46:41.2626436Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000003486)
2021-11-29T21:46:45.1531434Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000014583)
2021-11-29T21:46:45.1540589Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000014583)
2021-11-29T21:46:57.4351188Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000012287)
2021-11-29T21:46:57.4568349Z | FAILURE | BROKER | Consumer:
> Unable to write data to the transport connection: Broken pipe.
2021-11-29T21:47:01.4091516Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000012804)
2021-11-29T21:47:26.2146502Z | VERBOSE | CLIENT | Connected
2021-11-29T21:47:26.2150277Z | VERBOSE | SEND   | (1,255,[||])
2021-11-29T21:47:26.2494125Z | VERBOSE | RECV   | (15,32,[|0uy; …; 114uy|])
2021-11-29T21:47:26.2535665Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000012287)
2021-11-29T21:47:26.2546217Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000012804)
2021-11-29T21:47:38.5084953Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000009439)
2021-11-29T21:47:38.5090029Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000009439)
2021-11-29T21:47:42.4666772Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000005242)
2021-11-29T21:47:42.4671434Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000005242)

clipub.fsx

[nix-shell:~/code/dotnet/kafka/kaf]$ ./clipub.fsx 
2021-11-29T21:46:30.2332728Z | STARTED | CLIPUB | Poor Man's Kafka
2021-11-29T21:46:30.2333772Z | VERBOSE | CLIPUB | Press Enter to exit
2021-11-29T21:46:30.2386064Z | VERBOSE | SERVER | Connected
2021-11-29T21:46:30.2507960Z | VERBOSE | RECV   | (1,255,[||])
2021-11-29T21:46:30.2683352Z | VERBOSE | SEND   | (15,16,[|0uy; …; 114uy|])
2021-11-29T21:46:41.2523364Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000003486)
2021-11-29T21:46:45.1525000Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000014583)
2021-11-29T21:46:57.4346614Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000012287)
2021-11-29T21:47:01.4089258Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000012804)
2021-11-29T21:47:38.5080499Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000009439)
2021-11-29T21:47:42.4664506Z | VERBOSE | SEND   | Topic: foobar (bytes: 0000005242)

clisub.fsx

[nix-shell:~/code/dotnet/kafka/kaf]$ ./clisub.fsx 
2021-11-29T21:46:30.9806649Z | STARTED | CLISUB | Poor Man's Kafka
2021-11-29T21:46:30.9807929Z | VERBOSE | CLISUB | Press Enter to exit
2021-11-29T21:46:30.9853069Z | VERBOSE | SERVER | Connected
2021-11-29T21:46:30.9971614Z | VERBOSE | RECV   | (1,255,[||])
2021-11-29T21:46:31.0162422Z | VERBOSE | SEND   | (15,32,[|0uy; …; 114uy|])
2021-11-29T21:46:41.2676233Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000003486)
2021-11-29T21:46:45.1542172Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000014583)
2021-11-29T21:46:51.1911832Z | STOPPED | CLISUB | Poor Man's Kafka

[nix-shell:~/code/dotnet/kafka/kaf]$ ./clisub.fsx 
2021-11-29T21:47:26.2142101Z | STARTED | CLISUB | Poor Man's Kafka
2021-11-29T21:47:26.2143876Z | VERBOSE | CLISUB | Press Enter to exit
2021-11-29T21:47:26.2188991Z | VERBOSE | SERVER | Connected
2021-11-29T21:47:26.2310104Z | VERBOSE | RECV   | (1,255,[||])
2021-11-29T21:47:26.2499507Z | VERBOSE | SEND   | (15,32,[|0uy; …; 114uy|])
2021-11-29T21:47:26.2575274Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000012287)
2021-11-29T21:47:26.3647421Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000012804)
2021-11-29T21:47:38.5104967Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000009439)
2021-11-29T21:47:42.4672845Z | VERBOSE | RECV   | Topic: foobar (bytes: 0000005242)

create.fsx

[nix-shell:~/code/dotnet/kafka/kaf]$ for i in {1..2}; do ./create.fsx; done
2021-11-29T21:46:41.0342130Z | STARTED | EVENTS | Poor Man's Kafka
2021-11-29T21:46:41.1228100Z | CREATED | EVENTS | Random events: 4
2021-11-29T21:46:41.1297654Z | STOPPED | EVENTS | Poor Man's Kafka
2021-11-29T21:46:44.9934548Z | STARTED | EVENTS | Poor Man's Kafka
2021-11-29T21:46:45.0834605Z | CREATED | EVENTS | Random events: 22
2021-11-29T21:46:45.1482215Z | STOPPED | EVENTS | Poor Man's Kafka

[nix-shell:~/code/dotnet/kafka/kaf]$ for i in {1..2}; do ./create.fsx; done
2021-11-29T21:46:57.3332173Z | STARTED | EVENTS | Poor Man's Kafka
2021-11-29T21:46:57.4226179Z | CREATED | EVENTS | Random events: 17
2021-11-29T21:46:57.4296895Z | STOPPED | EVENTS | Poor Man's Kafka
2021-11-29T21:47:01.3094421Z | STARTED | EVENTS | Poor Man's Kafka
2021-11-29T21:47:01.4008672Z | CREATED | EVENTS | Random events: 16
2021-11-29T21:47:01.4080771Z | STOPPED | EVENTS | Poor Man's Kafka

[nix-shell:~/code/dotnet/kafka/kaf]$ for i in {1..2}; do ./create.fsx; done
2021-11-29T21:47:38.4121682Z | STARTED | EVENTS | Poor Man's Kafka
2021-11-29T21:47:38.5005655Z | CREATED | EVENTS | Random events: 15
2021-11-29T21:47:38.5071859Z | STOPPED | EVENTS | Poor Man's Kafka
2021-11-29T21:47:42.3704707Z | STARTED | EVENTS | Poor Man's Kafka
2021-11-29T21:47:42.4589979Z | CREATED | EVENTS | Random events: 9
2021-11-29T21:47:42.4663703Z | STOPPED | EVENTS | Poor Man's Kafka

Conclusion

It’s worth mentioning that this code has been produced over the weekend (at most 72 hours) when I noticed (the deadline) that I had committed to write this blog post for the first of December. It’s also worth mentioning, that the last many years, I haven’t been able to make FsAutoComplete work for emacs on neither of my *nix boxes and now on my macOS. So I have actually written all this code without any intellisense. Just so win people understand, it’s like starting notepad.exe and just typing code in it 😅.

Nevertheless, if we recall the key elements of Apache Kafka:

Service bus using stream-processing
High-throughput, low-latency platform for handling real-time data feeds
Uses a binary TCP-based protocol that is optimized for efficiency
Groups messages together to reduce the overhead of the network roundtrip
Larger: network packets, sequential disk operations & contiguous memory blocks

It should be clear to see that the mentioned key elements, are all part of this implementation, even though it might not be production ready.

Finally, I hope you have enjoyed the reading of this blog post. I guess the last thing to say is: Felices fiestas !!!

2021-12-01-poor-mans-kafka-stream-processing-bus_plata-o-plomo.jpg Cheers / skål / salud or «plata o plomo»?

Appendix: Source Code

common.fs

namespace FsAdvent

(* This construct is for ML compatibility. The syntax '(typ,...,typ) ident'
   is not used in F# code. Consider using 'ident<typ,...,typ>' instead. *)
#nowarn "62"

[<RequireQualifiedAccess>]
module Error =
  
  type error =
    { message    : string
    ; hresult    : int
    ; stacktrace : string
    ; inner      : error option
    }
  
  type t = private E of error with
    override t.ToString() = 
      let (E x) = t
      sprintf "%A" x
  
  let rec exn2error : System.Exception -> t =
    fun ex ->
      E
        { message    = ex.Message
        ; hresult    = ex.HResult
        ; stacktrace = ex.StackTrace
        ; inner      =
            if null = ex.InnerException 
            then
              None 
            else
              exn2error ex.InnerException
              |> function (E iex) -> Some iex
        }
  
  let message    (E x) = x.message
  let hresult    (E x) = x.hresult
  let stacktrace (E x) = x.stacktrace
  let inner      (E x) = x.inner

[<RequireQualifiedAccess>]
module Date =
  
  open System
  
  let timestamp : _ -> string =
    fun _ ->
      DateTime.UtcNow.ToString("o")

[<RequireQualifiedAccess>]
module Input =
  
  open System
  
  let wait : ConsoleKey -> unit =
    fun key ->
      let rec aux _ =
        match System.Console.ReadKey(true).Key with
          | k when k = key ->     ()
          | ______________ -> aux ()
      aux ()

[<RequireQualifiedAccess>]
module Output =
  
  open System
  
  let stdout : 'a -> unit =
    Console.WriteLine
  
  let stderr : 'a -> unit =
    fun x ->
      Console.Error.WriteLine x
  
  let debug : 'a -> unit =
    Diagnostics.Debug.WriteLine
  
  let trace : 'a -> unit =
    Diagnostics.Trace.WriteLine

[<RequireQualifiedAccess>]
module UID =
  
  open System
  
  let generate : unit -> Guid =
    Guid.NewGuid

[<RequireQualifiedAccess>]
module TCP =
    
  open System
  open System.IO
  open System.Net
  open System.Net.Sockets
  open System.Text
  open System.Threading 
  
  type communication = SEND | RECV
  
  type socket = private Config of config with
    interface IDisposable with
      member client.Dispose () =
        let (Config c) = client
        try
          c.socket.Shutdown(SocketShutdown.Both)
          c.socket.Close()
          c.cts.Cancel()
        with ex ->
          sprintf "%s | FAILURE | TCP Socket Dispose:\n%A" (Date.timestamp 0) ex
          |> Output.stderr
  
  and  config =
    { socket : Socket
    ; ip     : string
    ; port   : uint16
    ; cons   : int
    ; cts    : CancellationTokenSource
    }
  
  let private takeOwnershipSocket =
    (* Remark:
    
       The NetworkStream is created with read/write access to the specified
       Socket. If the value of ownsSocket parameter is true, the NetworkStream
       takes ownership of the underlying Socket, and calling the Close method
       also closes the underlying Socket.
       
       https://msdn.microsoft.com/
               en-us/library/te7e60bx(v=vs.110).aspx#Anchor_2
    *)
    false (* Taking ownership will dispose main socket *)
  
  let readAsync
    :  NetworkStream
    -> int
    -> byte array Async =
      fun stream count ->
        async {
          let  buf = Array.create count 0uy
          let! len =
            stream.AsyncRead
              ( buf
              , 0
              , count
              )
        
          if len = 0 then
            return [| |]
          else
            return buf[..len-1]
        }
  
  let writeAsync
    :  NetworkStream
    -> byte array
    -> unit Async =
      fun stream bytes ->
        async {
          return! stream.AsyncWrite bytes
        }
  
  let stream
    :  Socket
    -> NetworkStream =
      fun socket ->
        new NetworkStream
          ( socket
          , takeOwnershipSocket
          )
  
  let socket
    :  unit
    -> Socket =
      fun _ ->
        new Socket
          ( AddressFamily.InterNetwork
          , SocketType.Stream
          , ProtocolType.Tcp
          )
  
  let private exit =
    async {
      return
        ConsoleKey.Enter
        |> Input.wait
        |> Some
    }
  
  [<RequireQualifiedAccess>]
  module Server =
    
    type server = socket
    
    let init
      :  string
      -> int
      -> int
      -> server =
        fun ipaddress port cons ->
          let soc = socket ()
          (* Disable the Nagle Algorithm for this socket *)
          soc.NoDelay           <- true
          (* Set the receive buffer size to (2^31 - 1) *)
          soc.ReceiveBufferSize <- Int32.MaxValue
          (* Set the sender buffer size to (2^31 - 1) *)
          soc.SendBufferSize    <- Int32.MaxValue
          IPEndPoint
            ( IPAddress.Parse ipaddress
            , port
            )
          |> soc.Bind
          cons
          |> soc.Listen
          Config
            { socket = soc
            ; ip     = ipaddress
            ; port   = uint16 port
            ; cons   = cons
            ; cts    = new CancellationTokenSource()
            }
    
    let rec private loop server app =
      async {
        let (Config c) = server
        do
          Async.Start
            ( c.socket.Accept()
              |> app
            , c.cts.Token
            )
        return! loop server app
      }
    
    let listen
      :  server
      -> (Socket -> unit Async)
      -> unit option Async =
      fun server app ->
        [ exit
        ; loop server app
        ]
        |> Async.Choice
  
  [<RequireQualifiedAccess>]
  module Client =
    
    type client = socket
    
    let connect
      :  string
      -> int
      -> client =
        fun ipaddress port ->
          let soc = socket ()
          (* Disable the Nagle Algorithm for this socket *)
          soc.NoDelay           <- true
          (* Set the receive buffer size to (2^31 - 1) *)
          soc.ReceiveBufferSize <- Int32.MaxValue
          (* Set the sender buffer size to (2^31 - 1) *)
          soc.SendBufferSize    <- Int32.MaxValue
          IPEndPoint
            ( IPAddress.Parse ipaddress
            , port
            )
          |> soc.Connect
          Config
            { socket = soc
            ; ip     = ipaddress
            ; port   = uint16 port
            ; cons   = 1
            ; cts    = new CancellationTokenSource()
            }
    
    let private some client app =
      async {
        let (Config c) = client
        do! app c.socket
        return Some ()
      }
    
    let interact
      :  client
      -> (Socket -> unit Async)
      -> unit option Async =
      fun client app ->
        [ exit
        ; some client app
        ]
        |> Async.Choice

module Kafka =
  
  open System
  
  let inline (=>) x y = x, y
  
  module Protocol =
    
    open System.Text
    
    type data =
      | Octet of octet
      | Count of count
      | Bytes of bytes
      | Index of index
      | Topic of topic
      | Batch of batch
    and [<Struct>] octet = private Octet of byte
    and [<Struct>] count = private Count of int32
    and [<Struct>] bytes = private Bytes of byte array
    and [<Struct>] index = private Index of int32
    and            topic = private Topic of count * index * utf8:string
    and            batch = private Batch of topic *         msgs:bytes
    
    module Match =
      
      let (|Octet|) : octet -> byte                        = function
        | Octet x -> x
      let (|Count|) : count -> int32                       = function
        | Count x -> x
      let (|Bytes|) : bytes -> byte array                  = function
        | Bytes x -> x
      let (|Index|) : index -> int32                       = function
        | Index x -> x
      let (|Topic|) : topic -> int32 * string              = function
        | Topic (Count _, Index idx, utf8) ->
          ( idx
          , utf8
          )
      let (|Batch|) : batch -> int32 * string * byte array = function
        | Batch (Topic (idx, utf8), Bytes bs) ->
          ( idx
          , utf8
          , bs
          )
    
    [<RequireQualifiedAccess>]
    module Data =
      
      let octet
        :  byte
        -> data =
          Octet >> data.Octet
      
      let count
        :  int32
        -> data =
          Count >> data.Count
      
      let bytes
        :  byte array
        -> data =
          Bytes >> data.Bytes
      
      let index
        :  int32
        -> data =
          Index >> data.Index
      
      let topic
        :  int32
        -> string
        -> data =
          fun idx str ->
            let len =
              str
              |> Encoding.UTF8.GetBytes
              |> Array.length
            ( Count len
            , Index idx
            ,       str
            )
            |> Topic
            |> data.Topic
      
      let batch
        :  int32
        -> string
        -> byte array
        -> data =
          fun idx str bs ->
            let top =
              let len =
                str
                |> Encoding.UTF8.GetBytes
                |> Array.length
              ( Count len
              , Index idx
              ,       str
              )
            ( Topic top
            , Bytes bs
            )
            |> Batch
            |> data.Batch
    
    [<RequireQualifiedAccess>]
    module Binary =
      
      let private littleEndian = BitConverter.IsLittleEndian
      
      let fromOctet
        :  octet
        -> byte array =
          fun (Octet x) ->
            [| x |]
            
      let fromCount
        :  count
        -> byte array =
          fun (Count x) ->
            let bs = BitConverter.GetBytes x
            if littleEndian then
              bs
              |> Array.rev
            else
              bs
      
      let fromBytes
        :  bytes
        -> byte array =
          fun (Bytes xs) ->
            xs
            
      let fromIndex
        :  index
        -> byte array =
          fun (Index x) ->
            let bs = BitConverter.GetBytes x
            if littleEndian then
              bs
              |> Array.rev
            else
              bs
      
      let fromTopic
        :  topic
        -> byte array =
          fun (Topic (count, index, utf8)) ->
            seq {
              yield fromCount count
              yield fromIndex index
              yield Encoding.UTF8.GetBytes utf8
            }
            |> Array.concat
      
      let fromBatch
        :  batch
        -> byte array =
          fun (Batch(topic, bs)) ->
            Array.append (fromTopic topic) (fromBytes bs)
      
      let fromData
        :  data
        -> byte array =
          function
            | data.Octet x -> x |> fromOctet
            | data.Count x -> x |> fromCount
            | data.Bytes x -> x |> fromBytes
            | data.Index x -> x |> fromIndex
            | data.Topic x -> x |> fromTopic
            | data.Batch x -> x |> fromBatch
      
      let length
        :  data
        -> int32 =
          fromData >> Array.length
      
      let toOctet
        :  byte array
        -> octet =
          fun bs ->
            bs[0]
            |> Octet
      
      let private toInt32
        :  byte array
        -> int32 =
        fun bs ->
          let sof = 4
          let len = sof-1
          let bs' =
            if littleEndian then
              bs[..len]
              |> Array.rev
            else
              bs[..len]
          ( bs'
          , 0
          )
          |> BitConverter.ToInt32
      
      let toCount
        :  byte array
        -> count =
          toInt32 >> Count
      
      let toBytes
        :  byte array
        -> bytes =
          Bytes
      
      let toIndex
        :  byte array
        -> index =
          toInt32 >> Index
      
      let toTopic
        :  byte array
        -> topic =
        fun bs ->
          let sof = 4
          let cnt = toInt32                 bs[0*sof..1*sof-1]
          let idx = toInt32                 bs[1*sof..2*sof-1]
          let len = 2*sof+cnt
          let str = Encoding.UTF8.GetString bs[2*sof..  len-1]
          ( Count cnt
          , Index idx
          ,       str
          )
          |> Topic
      
      let toBatch
        :  byte array
        -> batch =
        fun bs ->
          let (off, top) =
            let sof = 4
            let cnt = toInt32                 bs[0*sof..1*sof-1]
            let idx = toInt32                 bs[1*sof..2*sof-1]
            let len = 2*sof+cnt
            let str = Encoding.UTF8.GetString bs[2*sof..  len-1]
            ( len
            , ( Count cnt
              , Index idx
              ,       str
              )
            )
          ( Topic top
          , Bytes bs[off..]
          )
          |> Batch
    
    type packet =
      private Packet of
        len : count *
        pid : octet *
        msg : bytes
      with
        override x.ToString () =
          let (Packet(Count len, Octet pid, Bytes msg)) = x
          sprintf "(%i,%i,%A)" len pid msg
    
    [<RequireQualifiedAccess>]
    module Packet =
      
      let init
        :  byte
        -> (string * data) list
        -> packet =
          fun pid ts ->
            let bs =
              ts
              |> List.map snd
              |> List.toArray
              |> Array.collect Binary.fromData
            let len =
              Array.length bs + 1
            Packet
              ( Count len
              , Octet pid
              , Bytes bs
              )
      
      let length
        :  packet
        -> int32 =
          fun (Packet(Count x, Octet _, Bytes _)) -> x
      
      let pid
        :  packet
        -> byte =
          fun (Packet(Count _, Octet x, Bytes _)) -> x
      
      let msg
        :  packet
        -> byte array =
          fun (Packet(Count _, Octet _, Bytes x)) -> x
      
      let push
        :  (byte array -> unit Async)
        -> packet
        -> unit Async =
          fun write (Packet(len, pid, msg)) ->
            async {
              do! len |> data.Count |> Binary.fromData |> write
              do! pid |> data.Octet |> Binary.fromData |> write
              do! msg |> data.Bytes |> Binary.fromData |> write
            }
      
      let pull
        :  (int -> byte array Async)
        -> packet Async =
          fun read ->
            async {
              let! lbs = read 4
              let! pbs = read 1
              let (len, len') =
                let (Count l') as l = Binary.toCount lbs
                ( l
                , l' - 1
                )
              let  pid = Binary.toOctet pbs
              let  mbs =
                async {
                  if 0 < len' then
                    let! tbs = read len'
                    return Binary.toBytes tbs
                  else
                    return Binary.toBytes [||]
                }
              let! msg = mbs
              
              return
                Packet
                  ( len
                  , pid
                  , msg
                  )
            }

broker.fsx

#!/usr/bin/env -S dotnet fsi --mlcompatibility --optimize --warnaserror+:25,26

(* This construct is for ML compatibility. The syntax '(typ,...,typ) ident'
   is not used in F# code. Consider using 'ident<typ,...,typ>' instead. *)
#nowarn "62"

#load "common.fs"

open System
open System.IO
open System.Linq
open System.Net.Sockets
open System.Text

open FsAdvent
open FsAdvent.Kafka
open FsAdvent.Kafka.Protocol.Match

let rec app
  :  Socket
  -> unit Async =
    fun socket ->
      async {
        Date.timestamp 0
        |> sprintf "%s | VERBOSE | CLIENT | Connected"
        |> Output.stdout
        
        let stream = TCP.stream socket
        
        (* Flow *)
        (* 0) Send welcome PID: 0xFFuy with no data *)
        let welcome =
          Protocol.Packet.init 0xFFuy
            [ "Welcome" => Protocol.Data.bytes [||]
            ]
        do! Protocol.Packet.push (TCP.writeAsync stream) welcome
        
        sprintf "%s | VERBOSE | %A   | %A"
          ( Date.timestamp 0 )
            TCP.SEND
            welcome
        |> Output.stdout
        
        (* 1) Recieve topic PID: 0x10uy with index and UTF-8 data *)
        let! topic = Protocol.Packet.pull (TCP.readAsync stream)
        
        sprintf "%s | VERBOSE | %A   | %A"
          ( Date.timestamp 0 )
            TCP.RECV
            topic
        |> Output.stdout
        
        match Protocol.Packet.pid topic with
          | 0x10uy (* Producer *) ->
            let top =
              match
                topic
                |> Protocol.Packet.msg
                |> Protocol.Binary.toTopic
              with
                |  Topic (_,top) -> top
            let dir =
              Path.Combine
                ( __SOURCE_DIRECTORY__
                , "topics"
                , top
                )
            (* Ensure folder exists *)
            Directory.CreateDirectory dir
            |> ignore
            
            return! producer socket top
          | 0x20uy (* Consumer *) ->
            let (idx,top) =
              match
                topic
                |> Protocol.Packet.msg
                |> Protocol.Binary.toTopic
              with
                |  Topic (idx,top) -> idx, top
            let dir =
              Path.Combine
                ( __SOURCE_DIRECTORY__
                , "topics"
                , top
                )
            (* Ensure folder exists *)
            Directory.CreateDirectory dir
            |> ignore
            
            return! consumer socket idx top
          | ______ ->
            Date.timestamp 0
            |> sprintf "%s | VERBOSE | CLIENT | Invalid Topic PID"
            |> Output.stdout
            socket.Shutdown(SocketShutdown.Both)
            socket.Close()
      }

and producer
  :  Socket
  -> string
  -> unit Async =
    fun socket topic ->
      async {
        try
          let stream = TCP.stream socket
        
          let! pkg = Protocol.Packet.pull (TCP.readAsync stream)
          
          (* Ensure unique filepath cross-producers *)
          let fp =
            Path.Combine
              ( __SOURCE_DIRECTORY__
              , "topics"
              , topic
              , ( Date.timestamp 0
                , UID.generate ()
                )
                ||> sprintf "%s_%A"
              )
          
          let bs =
            match
              pkg
              |> Protocol.Packet.msg
              |> Protocol.Binary.toBatch
            with
              |  Batch (_,_,bs) -> bs
          
          sprintf "%s | VERBOSE | %A   | Topic: %s (bytes: %010i)"
            ( Date.timestamp 0 )
              TCP.RECV
              topic
            ( bs |> Array.length )
          |> Output.stdout
        
          File.WriteAllBytes
            ( fp
            , bs
            )
          
          return! producer socket topic
        with ex ->
          sprintf "%s | FAILURE | BROKER | Producer:\n> %s"
            ( Date.timestamp 0 )
            ( ex |> Error.exn2error |> Error.message )
          |> Output.stderr
          socket.Shutdown(SocketShutdown.Both)
          socket.Close()
      }

and consumer
  :  Socket
  -> int32
  -> string
  -> unit Async =
    fun socket index topic ->
      async {
        try
          let stream = TCP.stream socket
          
          let dir =
            Path.Combine
              ( __SOURCE_DIRECTORY__
              , "topics"
              , topic
              )
        
          (* Flow: *)
          (* 0) Initially serve all batches from client index *)
          Directory
            .EnumerateFiles(dir)
            .Skip(index)
          |> Seq.sortBy id
          |> Seq.iter (
            fun x ->
              let bs  = File.ReadAllBytes x
              let pkg =
                Protocol.Packet.init 0xFFuy
                  [ "Batch" => Protocol.Data.batch 0 topic bs
                  ]
              
              async {
                do! Protocol.Packet.push (TCP.writeAsync stream) pkg
              }
              |> Async.RunSynchronously
              
              sprintf "%s | VERBOSE | %A   | Topic: %s (bytes: %010i)"
                ( Date.timestamp 0 )
                  TCP.SEND
                  topic
                ( bs |> Array.length )
              |> Output.stdout
          )
          
          (* 1) Listen to new batches added to the topic *)
          let fsw = new FileSystemWatcher(dir)
          fsw.IncludeSubdirectories <- false
          fsw.EnableRaisingEvents   <- true
          
          return! created socket topic fsw
        with ex ->
          sprintf "%s | FAILURE | BROKER | Consumer:\n> %s"
            ( Date.timestamp 0 )
            ( ex |> Error.exn2error |> Error.message )
          |> Output.stderr
          socket.Shutdown(SocketShutdown.Both)
          socket.Close()
      }

and created
  :  Socket
  -> string
  -> FileSystemWatcher
  -> unit Async =
    fun socket topic watch ->
      async {
        try
          let stream = TCP.stream socket
          
          let! e = Async.AwaitEvent watch.Created
          let bs =
            e.FullPath
            |> File.ReadAllBytes 
          let pkg =
            Protocol.Packet.init 0xFFuy
              [ "Batch" => Protocol.Data.batch 0 topic bs
              ]
        
          do! Protocol.Packet.push (TCP.writeAsync stream) pkg
          
          sprintf "%s | VERBOSE | %A   | Topic: %s (bytes: %010i)"
            ( Date.timestamp 0 )
              TCP.SEND
              topic
            ( bs |> Array.length )
          |> Output.stdout
          
          return! created socket topic watch
        with ex ->
          sprintf "%s | FAILURE | BROKER | Consumer:\n> %s"
            ( Date.timestamp 0 )
            ( ex |> Error.exn2error |> Error.message )
          |> Output.stderr
          socket.Shutdown(SocketShutdown.Both)
          socket.Close()
      }

let _ =
  try
    use server = TCP.Server.init "127.0.0.1" 22_222 10
    
    Date.timestamp 0
    |> sprintf "%s | STARTED | BROKER | Poor Man's Kafka"
    |> Output.stdout
    
    Date.timestamp 0
    |> sprintf "%s | VERBOSE | BROKER | Press Enter to exit"
    |> Output.stdout
    
    app
    |> TCP.Server.listen server
    |> Async.RunSynchronously
    |> ignore
    
    Date.timestamp 0
    |> sprintf "%s | STOPPED | BROKER | Poor Man's Kafka"
    |> Output.stdout
    00
  with ex ->
    ( Date.timestamp 0
    , ex |> Error.exn2error
    )
    ||> sprintf "%s | FAILURE | BROKER | Poor Man's Kafka\n%A"
    |> Output.stderr
    -1

domain.fs

namespace FsAdvent

(* This construct is for ML compatibility. The syntax '(typ,...,typ) ident'
   is not used in F# code. Consider using 'ident<typ,...,typ>' instead. *)
#nowarn "62"

module Domain =
  
  open System
  
  type guid = Guid
  type date = DateTime
  
  type FooBar =
    { uid : guid
    ; dts : date
    ; raw : char array
    }

[<RequireQualifiedAccess>]
module Random =
  
  open System
  
  open Domain
  
  let private foobar _ =
    let r = new Random ()
    let d =
      r.Next(000,900)
      |> double
    let n = r.Next(000,128)
    { uid = Guid.NewGuid ()
    ; dts = DateTime.UtcNow.AddSeconds(d * -1.0)
    ; raw =
      Array.init n (
        fun _ ->
          r.Next(047,127)
          |> byte
          |> char
      )
    }
  
  let foobars n =
    Seq.init n foobar

clipub.fsx

#!/usr/bin/env -S dotnet fsi --mlcompatibility --optimize --warnaserror+:25,26

(* This construct is for ML compatibility. The syntax '(typ,...,typ) ident'
   is not used in F# code. Consider using 'ident<typ,...,typ>' instead. *)
#nowarn "62"

#r "nuget: Newtonsoft.Json.Bson,  1.0.2"
#r "nuget: Newtonsoft.Json     , 13.0.1"

#load "common.fs"
#load "domain.fs"

open System
open System.IO
open System.Net.Sockets
open System.Text

open Newtonsoft.Json
open Newtonsoft.Json.Bson

open FsAdvent
open FsAdvent.Kafka
open FsAdvent.Kafka.Protocol.Match

let json<'a>
  :  string
  -> 'a =
    fun json ->
      try
        JsonConvert.DeserializeObject<'a>(json)
      with ex ->
        failwith ex.Message

let bson
  :  'a
  -> byte array =
    fun value ->
      use m = new MemoryStream  ( )
      use w = new BsonDataWriter(m)
      let s = new JsonSerializer( )
      s.Serialize(w, value)
      m.ToArray()

let rec app
  :  Socket
  -> unit Async =
    fun socket ->
      async {
        Date.timestamp 0
        |> sprintf "%s | VERBOSE | SERVER | Connected"
        |> Output.stdout
        
        let stream = TCP.stream socket
        
        (* Flow *)
        (* 0) Recieve welcome PID: 0xFFuy with no data *)
        let! welcome = Protocol.Packet.pull (TCP.readAsync stream)
        
        sprintf "%s | VERBOSE | %A   | %A"
          ( Date.timestamp 0 )
            TCP.RECV
            welcome
        |> Output.stdout
        
        (* 1) Send topic PID: 0x10uy with index and UTF-8 data *)
        let label = "foobar"
        let topic =
          Protocol.Packet.init (* Producer *) 0x10uy
            [ "Topic" => Protocol.Data.topic 0 label
            ]
        do! Protocol.Packet.push (TCP.writeAsync stream) topic
        
        sprintf "%s | VERBOSE | %A   | %A"
          ( Date.timestamp 0 )
            TCP.SEND
            topic
        |> Output.stdout
        
        (* Ensure folder exists *)
        let dir =
          Path.Combine
            ( __SOURCE_DIRECTORY__
            , "events"
            , label
            )
        dir
        |> Directory.CreateDirectory
        |> ignore
        
        (* Listen to new events added to the topic *)
        let fsw = new FileSystemWatcher(dir)
        fsw.IncludeSubdirectories <- false
        fsw.EnableRaisingEvents   <- true
        
        return! producer socket label fsw
      }

and producer 
  :  Socket
  -> string
  -> FileSystemWatcher
  -> unit Async =
    fun socket topic watch ->
      async {
        try
          let stream = TCP.stream socket
          
          let! e = Async.AwaitEvent watch.Created
        
          let bs =
            File.ReadAllText
              ( e.FullPath
              , Encoding.UTF8
              )
            |> json<Domain.FooBar array>
            |> bson
          let pkg =
            Protocol.Packet.init 0xFFuy
              [ "Batch" => Protocol.Data.batch 0 topic bs
              ]
          
          do! Protocol.Packet.push (TCP.writeAsync stream) pkg
          
          sprintf "%s | VERBOSE | %A   | Topic: %s (bytes: %010i)"
            ( Date.timestamp 0 )
              TCP.SEND
              topic
            ( bs |> Array.length )
          |> Output.stdout
          
          return! producer socket topic watch
        with ex ->
          sprintf "%s | FAILURE | CLIPUB | Producer:\n> %s"
            ( Date.timestamp 0 )
            ( ex |> Error.exn2error |> Error.message )
          |> Output.stderr
          socket.Shutdown(SocketShutdown.Both)
          socket.Close()
      }

let _ =
  try
    use clipub = TCP.Client.connect "127.0.0.1" 22_222
    
    Date.timestamp 0
    |> sprintf "%s | STARTED | CLIPUB | Poor Man's Kafka"
    |> Output.stdout
    
    Date.timestamp 0
    |> sprintf "%s | VERBOSE | CLIPUB | Press Enter to exit"
    |> Output.stdout
    
    app
    |> TCP.Client.interact clipub
    |> Async.RunSynchronously
    |> ignore
    
    Date.timestamp 0
    |> sprintf "%s | STOPPED | CLIPUB | Poor Man's Kafka"
    |> Output.stdout
    00
  with ex ->
    ( Date.timestamp 0
    , ex |> Error.exn2error
    )
    ||> sprintf "%s | FAILURE | CLIPUB | Poor Man's Kafka\n%A"
    |> Output.stderr
    -1

clisub.fsx

#!/usr/bin/env -S dotnet fsi --mlcompatibility --optimize --warnaserror+:25,26

(* This construct is for ML compatibility. The syntax '(typ,...,typ) ident'
   is not used in F# code. Consider using 'ident<typ,...,typ>' instead. *)
#nowarn "62"

#r "nuget: Newtonsoft.Json.Bson,  1.0.2"
#r "nuget: Newtonsoft.Json     , 13.0.1"

#load "common.fs"
#load "domain.fs"

open System
open System.IO
open System.Net.Sockets
open System.Text

open Newtonsoft.Json
open Newtonsoft.Json.Bson

open FsAdvent
open FsAdvent.Kafka
open FsAdvent.Kafka.Protocol.Match

let bson<'a>
  :  byte array
  -> 'a =
    fun a ->
      use m = new MemoryStream  (a)
      use r = new BsonDataReader(m)
      let s = new JsonSerializer( )
      r.ReadRootValueAsArray <- true
      s.Deserialize<'a>(r)

let json
  :  'a
  -> string =
    fun value ->
      let settings = new JsonSerializerSettings()
      settings.DefaultValueHandling <- DefaultValueHandling.Include
      JsonConvert.SerializeObject
        ( value
        , Formatting.Indented
        , settings
        )

let rec app
  :  Guid
  -> Socket
  -> unit Async =
    fun uid socket ->
      async {
        Date.timestamp 0
        |> sprintf "%s | VERBOSE | SERVER | Connected"
        |> Output.stdout
        
        let stream = TCP.stream socket
        
        (* Flow *)
        (* 0) Recieve welcome PID: 0xFFuy with no data *)
        let! welcome = Protocol.Packet.pull (TCP.readAsync stream)
        
        sprintf "%s | VERBOSE | %A   | %A"
          ( Date.timestamp 0 )
            TCP.RECV
            welcome
        |> Output.stdout
        
        (* 1) Send topic PID: 0x20uy with index and UTF-8 data *)
        (* Ensure folder exists *)
        let label = "foobar"
        let state =
          let dir =
            Path.Combine
              ( __SOURCE_DIRECTORY__
              , "client"
              , sprintf "%A" uid
              , "states"
              , label
              )
          dir
          |> Directory.CreateDirectory
          |> ignore
          
          Directory.EnumerateFiles(dir)
          |> Seq.length
        let topic =
          Protocol.Packet.init (* Consumer *) 0x20uy
            [ "Topic" => Protocol.Data.topic state label
            ]
        do! Protocol.Packet.push (TCP.writeAsync stream) topic
        
        sprintf "%s | VERBOSE | %A   | %A"
          ( Date.timestamp 0 )
            TCP.SEND
            topic
        |> Output.stdout
        
        (* Ensure folder exists *)
        Path.Combine
          ( __SOURCE_DIRECTORY__
          , "client"
          , sprintf "%A" uid
          , "stream"
          , label
          )
        |> Directory.CreateDirectory
        |> ignore
        
        return! consumer socket uid label
      }

and consumer 
  :  Socket
  -> Guid
  -> string
  -> unit Async =
    fun socket uid topic ->
      async {
        try
          let stream = TCP.stream socket
          
          let! pkg = Protocol.Packet.pull (TCP.readAsync stream)
          
          let fp sub =
            Path.Combine
              ( __SOURCE_DIRECTORY__
              , "client"
              , sprintf "%A" uid
              , sub
              , topic
              , ( Date.timestamp 0
                , UID.generate ()
                )
                ||> sprintf "%s_%A"
              )
          
          let bs =
            match
              pkg
              |> Protocol.Packet.msg
              |> Protocol.Binary.toBatch
            with
              |  Batch (_,_,bs) -> bs
          
          sprintf "%s | VERBOSE | %A   | Topic: %s (bytes: %010i)"
            ( Date.timestamp 0 )
              TCP.RECV
              topic
            ( bs |> Array.length )
          |> Output.stdout
          
          bs
          |> bson<Domain.FooBar array>
          |> Array.iter(
            fun x ->
              File.WriteAllText
                ( fp "stream"
                , json x
                , Encoding.UTF8
                )
          )
          
          File.WriteAllText
            ( fp "states"
            , String.Empty
            , Encoding.UTF8
            )
          
          return! consumer socket uid topic
        with ex ->
          sprintf "%s | FAILURE | BROKER | Consumer:\n> %s"
            ( Date.timestamp 0 )
            ( ex |> Error.exn2error |> Error.message )
          |> Output.stderr
          socket.Shutdown(SocketShutdown.Both)
          socket.Close()
      }

let _ =
  try
    use clisub = TCP.Client.connect "127.0.0.1" 22_222
    
    Date.timestamp 0
    |> sprintf "%s | STARTED | CLISUB | Poor Man's Kafka"
    |> Output.stdout
    
    Date.timestamp 0
    |> sprintf "%s | VERBOSE | CLISUB | Press Enter to exit"
    |> Output.stdout
    
    Guid.Empty (* Use `empty` for simplicity and determinism *)
    |> app
    |> TCP.Client.interact clisub
    |> Async.RunSynchronously
    |> ignore
    
    Date.timestamp 0
    |> sprintf "%s | STOPPED | CLISUB | Poor Man's Kafka"
    |> Output.stdout
    00
  with ex ->
    ( Date.timestamp 0
    , ex |> Error.exn2error
    )
    ||> sprintf "%s | FAILURE | CLISUB | Poor Man's Kafka\n%A"
    |> Output.stderr
    -1

create.fsx

#!/usr/bin/env -S dotnet fsi --mlcompatibility --optimize --warnaserror+:25,26

(* This construct is for ML compatibility. The syntax '(typ,...,typ) ident'
   is not used in F# code. Consider using 'ident<typ,...,typ>' instead. *)
#nowarn "62"

#r "nuget: Newtonsoft.Json, 13.0.1"

#load "common.fs"
#load "domain.fs"

open System
open System.IO
open System.Text

open Newtonsoft.Json

open FsAdvent

let json
  :  'a
  -> string =
    fun value ->
      let settings = new JsonSerializerSettings()
      settings.DefaultValueHandling <- DefaultValueHandling.Include
      JsonConvert.SerializeObject
        ( value
        , Formatting.None
        , settings
        )

let _ =
  try
    Date.timestamp 0
    |> sprintf "%s | STARTED | EVENTS | Poor Man's Kafka"
    |> Output.stdout
      
    let ran = new Random()
    let len = ran.Next(01,32)
    let lab = "foobar"
    (* Ensure folder exists *)
    Path.Combine
      ( __SOURCE_DIRECTORY__
      , "events"
      , lab
      )
    |> Directory.CreateDirectory
    |> ignore
    
    len
    |> Random.foobars
    |> json
    |> fun x ->
      sprintf "%s | CREATED | EVENTS | Random events: %i"
        ( Date.timestamp 0 )
          len
      |> Output.stdout
      
      let p =
        Path.Combine
          ( __SOURCE_DIRECTORY__
          , "events"
          , lab
          , ( Date.timestamp 0
            , UID.generate ()
            )
            ||> sprintf "%s_%A"
          )
      
      File.WriteAllText
        ( p
        , x
        , Encoding.UTF8
        )
    
    Date.timestamp 0
    |> sprintf "%s | STOPPED | EVENTS | Poor Man's Kafka"
    |> Output.stdout
    00
  with ex ->
    ( Date.timestamp 0
    , ex |> Error.exn2error
    )
    ||> sprintf "%s | FAILURE | EVENTS | Poor Man's Kafka\n%A"
    |> Output.stderr
    -1

remove.sh

#!/bin/sh

clear

rm -frv ./client/*
rm -frv ./events/*
rm -frv ./topics/*

F# Advent Calendar 2021

Background

Poor man’s Kafka from scratch

Prerequisites

Structure

broker.fsx

clipub.fsx

clisub.fsx

create.fsx

Conclusion

Appendix: Source Code

common.fs

broker.fsx

domain.fs

clipub.fsx

clisub.fsx

create.fsx

remove.sh

Recommend

China threatens retaliation against diplomatic boycott of Olympics: 'The US will...

Rust语言的核心开发团队有毒 - HackMD

事件驱动架构正在起飞 – Boyney.io

Harvard team disables newly discovered hormone to reverse diabetes in mice

Semantics, Not Syntax; Developer empowerment using functional-first programming

Evergrande has defaulted on its debt, Fitch Ratings says

又一个基于JVM的编程语言：Flix

私域，数字化时代的「金矿」

Mercedes-Benz gets world’s first approval for automated driving system

Setting up FNA and F# for game development

About Joyk