gab

Docs

Gabonomicon

Examples

Networked Kv Store

This example extends the in-memory key-value store with a TCP server. Clients connect and send text commands; the server parses each line, dispatches it to the store actor, and writes a response.

Protocol

Since we’re working over the wire, we need a protocol! Lets make it simple. Each request is a single newline-terminated line. Each response is a single newline-terminated line.

Request	Response	No key response
`GET key`	`OK value`	`NONE`
`SET key value`	`OK`	—
`DELETE key`	`OK value`	`NONE`

Handling a Connection

Lets add a message to gab’s socket type, called serve_client. We’ll send this message to each tcp client we accept from our tcp server.

Each accepted connection gets its own fiber. The fiber reads one line, dispatches to the store, writes the response, and recurses.

serve_client: .def (Io.Sockets.t, (store) :: do
  data := self.transduce(
      Binaries.make,
      (x y) :: x + y,
      Transducers.take_until(x :: x == '\n'.to\binary))

  line := data.as\string.unwrap

  (cmd, args*) := line.trim.to\stream\split(' ').collect [] .*

  response := cmd.to\message.run_command(self, store, args*)
  self.stream\send('$\n'.sprintf(response).to\binary)

  self.serve_client store
end)

We use a transducer to take bytes off the stream until we see a newline byte. We split the line we read by first converting it into a seqable with to\stream\split. This creates a seqable which will yield pieces of the string separated by the separator, in this case a space.

The command handlers receive the socket, the store, and any remaining arguments:

run_command: .defcase {
  GET: (socket, store, key, rest*) :: do
    store.store\get key
      .then((val) :: 'OK $'.sprintf(val))
      .else(()    :: 'NONE')
  end

  SET: (socket, store, key, val, rest*) :: do
    store.store\set (key val)
      .then(() :: 'OK')
  end

  DELETE: (socket, store, key, rest*) :: do
    store.store\delete key
      .then((val) :: 'OK $'.sprintf(val))
      .else(()    :: 'NONE')
  end
}

The defcase keys are GET:, SET:, and DELETE:. They explicitly match what to\m produces from the wire protocol strings. rest* absorbs any extra tokens so malformed commands don’t crash the handler. Missing arguments arrive as nil:, which the store returns none: for, propagating back to the client as NONE.

Accepting Clients

The server accepts connections one at a time, immediately spawning a fiber for each and looping:

store\handle_client: .defcase {
  ok: (client, store) :: do
    'Serving client $\n'.printf client
    Fibers.make(client serve_client: store).await
  end
  err: (msg) :: do
    'Server failed to accept client: $\n'.printf(msg)
  end
}

[Io.Sockets.t] .defmodule {
  store\accept_loop:
  (store) :: do
    self
      .accept
      .store\handle_client(store)

    self
      .store\accept_loop store
  end
}

server := self captures the socket before the then: block, where self would refer to the block. accept blocks until a client connects; a new fiber is spawned immediately and the loop recurses without waiting for that client to finish.

Starting the Server

start: is defined on the store type. It creates the socket, binds, listens, and launches the accept loop in a fiber:

[Store.t] .defmodule {
  store\listen: (host port) :: do
    server := Io.Sockets.make(tcp:).unwrap
    server.bind(host port).unwrap
    server.listen(128).unwrap
    'Listening on $:$'.sprintf(host port).println
    store := self
    Fibers.make () :: server.store\accept_loop(store)
  end
}

Inside the fiber, self would refer to the fiber’s own block. Because of this, we capture the self as store beforehand.

Putting it Together

Here is how you can now use the store:

store  := Store.make
server := store.start('::1' 6379)
server.await

Connect with any TCP client:

$ echo "SET name gab" | nc ::1 6379
OK

$ echo "GET name" | nc ::1 6379
OK gab

$ echo "DELETE name" | nc ::1 6379
OK gab

$ echo "GET name" | nc ::1 6379
NONE

The Full Program

The full store module now looks like this:

Stores := 'examples' .use 'kvstore'

store\handle_client: .defcase {
  ok: (client, store) :: do
    'Serving client $\n'.printf client
    Fibers.make(client serve_client: store).await
  end
  err: (msg) :: do
    'Server failed to accept client: $\n'.printf(msg)
  end
}

[Io.Sockets.t] .defmodule {
  store\accept_loop:
  (store) :: do
    self
      .accept
      .store\handle_client(store)

    self
      .store\accept_loop store
  end
}

[Stores.t] .defmodule {
  store\listen: (host port) :: do
    server := Io.Sockets.make(tcp:).unwrap
    server.bind(host port).unwrap
    server.listen(128).unwrap
    'Listening on $:$'.sprintf(host port).println
    store := self
    Fibers.make () :: server.store\accept_loop(store)
  end
}

run_command: .defcase {
  GET: (socket, store, key, rest*) :: do
    store.store\get key
      .then((val) :: 'OK $'.sprintf(val))
      .else(()    :: 'NONE')
  end

  SET: (socket, store, key, val, rest*) :: do
    store.store\set (key val)
      .then(() :: 'OK')
  end

  DELETE: (socket, store, key, rest*) :: do
    store.store\delete key
      .then((val) :: 'OK $'.sprintf(val))
      .else(()    :: 'NONE')
  end
}

serve_client: .def (Io.Sockets.t, (store) :: do
  sock := self

  data := sock.transduce(
      Binaries.make,
      (x y) :: x + y,
      Transducers.take_until(x :: x == '\n'.to\binary))

  line := data.as\string.unwrap

  (cmd, args*) := line.trim.to\stream\split(' ').collect [] .*

  response := cmd.to\message.run_command(sock, store, args*)
  sock.stream\send('$\n'.sprintf(response).to\binary)

  sock.serve_client store
end)

store  := Stores.make
server := store.store\listen('::1' 6379)
server.await

Kv Store Pub/Sub Broker