Skip to content

Commit

Permalink
Nextgen: Tristanable v2 (#4)
Browse files Browse the repository at this point in the history
* - Added new logo to `README.md`
- Fixed typos in `README.md`
- Added new logo (source included)

* - Added `bformat` version `3.1.13` as dependency

* - Removed executable

* - Updated `.gitignore`

* - Use `https` link rather to `bformat` homepage

* Package (tristanable)

- Added public imports along with comments per each

Encoding

- Added a stub class, `TaggedMessage`, for encoding and decoding the tristanable byte payload

Exceptions

- Added `TristanableException` exception type along with the `Error` enum sub-type

Manager

- Added stub code for `Manager` to manage the queues and socket

Queue

- Added stub class representing a queue with a tag (`Queue`)

QueueItem

- Added stub class `QueueItem` which represents an item that is enqueued/dequeued onto a `Queue`

Watcher

- Added stub class `Watcher` which will manage the socket reading-wise

* Manager

- Added field `watcher` of type `Watcher`

* Watcher

- Added constructor which takes in an instance of `Manager` and an instance of `Socket`

* Manager

- Added unit test TODO

* - Moved `Watcher` and `Manager` modules to their own package
- Ensured `Watcher`'s constructor is package-level accessible only

Manager

- The constructor now creates an instance of `Watcher`
- Added a `start()` method which calls `watcher.start()`

* Manager

- Added stub `sendMessage(TaggedMessage)` which will encode into the tristanable format, then wrap into bformat and send over the socket
- Added import for `TaggedMessage` from `tristanable.encoding` module

* Package (tristanable)

- Added an import for `TaggedMessage` from module `tristanable.encoding`

* Encoding

- Added stub class `TaggedMessage`
- Added constructor, static decoder (unimplemented), `encoder (implemented), getters and setters
- Added module `tristanable.encoding`

* - Attempt merge

* Encoding

- Added parameter-less (default) constructor marked as `private` to `TaggedMessage`
- Added decoding support in `decode(byte[])` which will return a new instance of `TaggedMessage`
- Added a unit test to test encoding and decoding

* TaggedMessage

- Added documentation for fields `tag` and `data`
- Added documentation for both constructors
- Added documentation for `getPayload()`, `getTag()`, `setPayload(byte[])` and `setTag(ulong)`

* - Fixed formatting in `README.md`

* Watcher

- Added import for `bformat` and `encoding` module
- Documented `watch()`
- Added `bformat` read-and-decode `receiveMessage(Socket, ref byte[])` call followed by a `TaggedMessage.decode(byte[])` call

* Watcher

- Moved TODO below already completed code

Exceptions

- Renamed `Error` to `ErrorType`
- Constructing a new `TristanableException` will now store the passed in `ErrorType`
- Added `getError()` to `TristanableException` which returns the stored `ErrorType`
- Added two new memebrs to enum `ErrorType`, namely `QUEUE_NOT_FOUND` and `QUEUE_ALREADY_EXISTS`

* Manager

- Changed from using D's dynamic arrays for the array of `Queue` objects to using an `SList!(T)` where `T` is the `Queue` type
- Implemented `getQueue(ulong)` which returns the `Queue` object with the matching id/tag, else throws an instance of `TristanabaleException`
- Implemented `registerQueue(Queue)` which will attempt to add the provided `Queue` given that a queue does not already exist with the provided queue's id; if that is the case then an instance of `TristanableException` is thrown

Queue

- Made the constructor take in the `ulong` queue ID
- Made the constructor publically accessible
- Implemented `getID()` which returns the `Queue`'s id as a `ulong`
- Removed the static method `newQueue(ulong)`

Unit test

- Added a unit test to test `getQueue(ulong)` when the queue cannot be found
- Added a unit test to test adding a queue and successfully retrieving it

* Package (tristanable)

- Fixed up the `exceptions` module import

* Manager

- Removed now-completed TODO in `registerQueue(QUeue)`

* Manager

- Documented `registerQueue(Queue)`

* Exceptions

- Added new member `NO_DEFAULT_QUEUE` to enum `ErrorType`

* Queue

- The actual queue is now an `SList!(TaggedMessage)`
- Added a stub method `enqueue(TaggedMessage)`
- Updated the stub method `dequeue()` which returns a `TaggedMessage` now

* - Removed `QueueItem`

* Encoding

- Implemented `toString()` in `TaggedMessage`

* Manager

- Added a default queue
- `getQueue(ulong)` now calls `getQueue_nothrow(ulong)` with the same id
- Implemented `getQueue_nothrow(ulong)` which returns the `Queue` if found, `null` otherwise
- Added `getDefaultQueue()` which gets the default queue by calling `getDefaultQueue_nothrow(ulong)` with the same id
- Added `getDefaultQueue_nothrow(ulong)` which returns the default queue as a `Queue` object if it exists, else `null`
- Added `setDefaultQueue(Queue)` which sets the provided queue as the default queue (i.e. the queue where messages tagged with a tag of a queue not registered will be dumped into - if the default queue is set)

Watcher

- Set the worker thread, `watch`, in the constructor
- Added a TODO relating to checking if the socket read succeeded or not
- Added a debug print for the received `TaggedMessage` post-decode
- Extract the tag of the message and find the matching queue (potentially, if it exists)
- If the queue exists then add the `TaggedMessage` to said `Queue`
- If the queue doesn't exist then, get the so-called "Default queue", if it doesn't exist don't do anything, if it does then enqueue the message (the `TaggedMessage`) to said `Queue`

Unit test

- Added a unit test (WIP) for testing the `Manager` and `Watcher` mechanism
- Updated unittest to test the `getQueue_nothrow(ulong)` method
- Added a unit test to test adding a `Queue` with a tag that already exists in a `Queue` registered prior

* Queue

- Added imports for `std.stdio` and `to` from `std.conv` to be imported when compiling in `unittest` mode
- Added documentation to `enqueue(TaggedMessage)`
- Implemented `enqueue(TaggedMessage)` using libsnooze
- Added documentation for `dequeue()`
- Implemented `dequeue()` using libsnooze

* Queue

- Replaced now-completed TODO with an actual comment in `dequeue()`

* Queue

- Added entrance and exit debugs for `dequeue()`

* Manager

- Clean up I guess

* Unit test (Watcher)

- Unit test for watcher works

* Watcher

- Deleted old module that was unused

* Package

- Removed completed TODO

* Exceptions

- Removed unused enum member `QueueExists` of enum `ErrorType`

* Encoding

- Added module-level documentation

Exceptions

- Added module-level documentation

Queue

- Added module-level documentation

Package (`tristanable.manager`)

- Added module-level documentation

* Manager

- Implemented `getUniqueQueue()` which finds an unused tag, makes a `Queue` with said tag, registers it and then returns it
- WIP: `shutdown()` method

* Manager

- Added unittest for `getUniqueQueue()`
- Typo fix

* Manager

- Removed empty unittest

* Config

- Added new module `config`
- Added new type `Config` which is used for configuring an instance of `Manager`
- Added `defaultConfig()` which returns the default `Config` instance used

* Config

- Make the default configuration generated more explicit in `defaultConfig`
- Remove initialization in `Config` (would be `false` in any case)

* Manager

- Added support for configuring the `Manager`, the constructor now uses the default configuration
- Implemented `registerQueue_nothrow(Queue)` which returns `true` on success, `false` otherwise
- `registerQueue(Queue)` now makes a sub-call to `registerQueue_nothrow(Queue)`
- Implemented `sendMessage(TaggedMessage)`
- Added comment for assertions in unittest

* Package (`manager`)

- Import the `Config` type and the `defaultConfig()` function

* Watcher

- Added stub `shutdown()` method that is intended to be called by `Manager` (package-level accessible)

Unit tests

- Sleep for 4 seconds instead of 2 before the server sends the two tagged messages
- Send a messae tagged with tag `42` before that of the one tagged with `69`
- Register a queue with tag `42`
- Remove `WaitingThread`, we now receive both tagged messages (`42` and `69`) on the unittest thread
- Added assertion to ensure the tagged message of `69` is indeed received correctly (tag AND payload)

* Queue

- Added documentation for the constructor `this(ulong)`
- Fixed issue #5

Unit tests

- The `==` operator on strings does some normalization stuff which results in differing byte sequences and therefore inequality (see the `"Cucumber 😳️"` case), therefore casting to `byte[]`
- Send another message tagged with `69`
- Fixed comment for server code sending tagged `42` message
- Call `manager.stop()` right at the end of the unit test

* Watcher

- Added package-level accessible `startWatcher()` method which calls `start()` for us
- Added some debugging prints which will now only be compiled-in during unittest builds
- If the bformat `receiveMessage(Socket, ref byte[])` method fails (returns `false`) then exit the loop, only continue decoding if it is `true`
- Implemented package-level accesible `shutdown()` method

Manager

- `start()` now calls `watcher.startWatcher()` instead of `watcher.start()`

* Unit test

- Sleep a little longer for profiling tests

* Watcher

- Documented unittest as it is a great example of how to sue tristanable

* Exceptions

- Documented `ErrorType` and all its members
- Documented `TristanableException`

* Exceptions

- Added missing documentation
- Fixed the message generation of the exception's message

* Manager

- Added documentation for `start()` and `stop()`

* Manager

- Removed now-completed TODO
- Added documentation for `queuesLock`

* Watcher

- Added documentation to the constructor for `Watcher`

* Watcher

- Documented module `tristanable.manager.watcher`

* Queue

- Docuemneted `getId()`
- Documented the `Queue` class
- Documented fields `event`, `queue` and `queueLock`

* Package

- Removed whitespace

* Watcher

- Documented method `shutdown()`

* - Upgraded to new `bformat` version `4.1.0` and migrated to using `BClient` (unit tests seem to pass)

* Manager

- Added a TODO for the future `removeQueue(Queue)` and `removeQueu_nothrow(Queue)`

* Manager

- Implemented `releaseQueue(Queue)`
- Implemented `releaseQueue_nothrow(Queue)`

Unit tests

- Added unit test for `releaseQueue(Queue)`

* Unit tests

- Added a TODO

* Dub

- Now requires a minimum version of `libsnooze` of at least `
1.0.0-beta`

* Queue

- Be specific, catch `FatalException` in `enqueue(TaggedMessage)`
- Be specific, catch `InterruptedException` and `FatalException` seperately

* Exceptions

- Added enum members `DEQUEUE_FAILED` and `ENQUEUE_FAILED` to `ErrorType` enum

* Queue

- `enqueue(TaggedMessage)` can now throw a `TristanableException` if a `FataException` with `libsnooze` occurrs
- `dequeue()` can now throw a `TristanableException` if a `FatalException` occurs during the call to `wait()` on `libsnooze`

* Dub

- Required minimum version of `bformat`, `4.1.0`

* Dub

- Upgraded `libsnooze` to version `1.3.0-beta`

* Migrate from libsnooze (#8)

* Dub

- Removed `libsnooze` dependency

* Queue

- Removed `libsnooze` imports

* Queue

- Added mutex+condition variable

* Queue

- Removed old `ensure()` call

* Queue

- Switched one thing over to mutex+condvar

* Queue

- Switched to using condition variable
- Added configurable slumber interval

* Queue

- Removed TODOs which are irrevevant for now

* Queue

- Removed `TListener` references

Everything else

- Removed reference to old/duplicate `queue.d` module

* Hotfix/niknaks (#9)

* Dub

- Added `niknaks` package with a minimum version of `v0.3.0`

* Encoding

- Switched to niknaks for `decode()`

* Encoding

- `encode()` now uses niknaks

* Watcher (unit tests)

- Added testing for default queue
  • Loading branch information
deavmi committed Oct 2, 2023
1 parent a23017a commit b24c336
Show file tree
Hide file tree
Showing 17 changed files with 1,408 additions and 81 deletions.
1 change: 1 addition & 0 deletions .gitignore
Expand Up @@ -24,3 +24,4 @@ docs/
*.lst
source/tristanable/queue.d
dub.selections.json
tristanable-test-library
18 changes: 9 additions & 9 deletions README.md
@@ -1,24 +1,24 @@
![](https://code.dlang.org/packages/tristanable/logo?s=5ef1c9f1250f57dd4c37efbf)
![](branding/logo_small.png)

tristanable
===========

[![D](https://github.com/deavmi/tristanable/actions/workflows/d.yml/badge.svg)](https://github.com/deavmi/tristanable/actions/workflows/d.yml) ![DUB](https://img.shields.io/dub/v/tristanable?color=%23c10000ff%20&style=flat-square) ![DUB](https://img.shields.io/dub/dt/tristanable?style=flat-square) ![DUB](https://img.shields.io/dub/l/tristanable?style=flat-square) [![Coverage Status](https://coveralls.io/repos/github/deavmi/tristanable/badge.svg?branch=master)](https://coveralls.io/github/deavmi/tristanable?branch=master)


**Tristanable** is a library for D-based libraries and applications that need a way to receive variable-length messages of different types (via a `Socket`) and place these messages into their own resepctively tagged queues indicated by their _"type"_ or `id`.
**Tristanable** is a library for D-based libraries and applications that need a way to receive variable-length messages of different types (via a `Socket`) and place these messages into their own respectively tagged queues indicated by their _"type"_ or `id`.

## What problems does it solve?

### Human example

Say now you made a request to a server with a tag `1` and expect a reply with that same tag `1`. Now, for a moment, think about what would happen in a tagless system. You would be expecting a reply, say now the weather report for your area, but what if the server has another thread that writes an instant messenger notification to the server's socket before the weather message is sent? Now you will inetrpret those bytes as if they were a weather message.
Say now you made a request to a server with a tag `1` and expect a reply with that same tag `1`. Now, for a moment, think about what would happen in a tagless system. You would be expecting a reply, say now the weather report for your area, but what if the server has another thread that writes an instant messenger notification to the server's socket before the weather message is sent? Now you will interpret those bytes as if they were a weather message.

Tristanable provides a way for you to receive the "IM notification first" but block and dequeue (when it arrives in the queue) for the "weather report". Irresepctoive of wether (no pun intended) the weather report arrives before the "IM notification" or after.
Tristanable provides a way for you to receive the "IM notification first" but block and dequeue (when it arrives in the queue) for the "weather report". Irrespective of wether (no pun intended) the weather report arrives before the "IM notification" or after.

### Code example

If we wanted to implement the following we would do the following. One note is that instead of waiting on messages of a specific _"type"_ (or rather **tag**), tristanable provides not just a one-message lengthb uffer per tag but infact a full queue per tag, meaning any received message with tag `1` will be enqueued and not dropped after the first message of type `1` is buffered.
If we wanted to implement the following we would do the following. One note is that instead of waiting on messages of a specific _"type"_ (or rather **tag**), tristanable provides not just a one-message length buffer per tag but in fact a full queue per tag, meaning any received message with tag `1` will be enqueued and not dropped after the first message of type `1` is buffered.

```d
import tristanable.manager;
Expand All @@ -42,7 +42,7 @@ manager.addQueue(instantNotification);
QueueItem message = weatherQueue.dequeue();
```

Surely, there must be some sort of encoding mechanism too? The messages afterall need to be encoded. **No problem!**, we have that sorted:
Surely, there must be some sort of encoding mechanism too? The messages after all need to be encoded. **No problem!**, we have that sorted:

```d
import tristanable.encoding;
Expand All @@ -58,7 +58,7 @@ DataMessage tristanEncoded = new DataMessage(tag, data);
socket.send(encodeForSend(tristanEncoded));
```

And let tristanable handle it! We even handle the message lengths and everything using another great project [bformat](http://deavmi.assigned.network/projects/bformat).
And let tristanable handle it! We even handle the message lengths and everything using another great project [bformat](https://deavmi.assigned.network/projects/bformat).

## Format

Expand All @@ -67,8 +67,8 @@ And let tristanable handle it! We even handle the message lengths and everything
```

## Using tristanable in your D project
You can easily add the library (source-based) to your project by running the following command in your
project's root:

You can easily add the library (source-based) to your project by running the following command in your project's root:

```bash
dub add tristanable
Expand Down
Binary file added branding/logo_base.xcf
Binary file not shown.
Binary file added branding/logo_small.png
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Binary file added branding/logo_small.xcf
Binary file not shown.
5 changes: 3 additions & 2 deletions dub.json
Expand Up @@ -2,12 +2,13 @@
"authors": [
"Tristan B. Velloza Kildaire"
],
"homepage": "https://deavmi.assigned.network/projects/tristanable",
"copyright": "Copyright © 2023, Tristan B. Kildaire",
"dependencies": {
"libsnooze": "0.3.3"
"bformat": ">=4.1.1",
"niknaks": ">=0.3.0"
},
"description": "Tristanable network message queuing framework",
"homepage": "https://deavmi.assigned.network/projects/tristanable",
"license": "LGPL-3.0",
"name": "tristanable",
"targetType": "library"
Expand Down
168 changes: 168 additions & 0 deletions source/tristanable/encoding.d
@@ -0,0 +1,168 @@
/**
* Encoding/decoding of the tristanable format
*/
module tristanable.encoding;

import std.conv : to;
import niknaks.bits : bytesToIntegral, Order, order, toBytes;

/**
* Represents a tagged message that has been decoded
* from its raw byte encoding, this is a tuple of
* a numeric tag and a byte array of payload data
*
* Also provides a static method to decode from such
* raw encoding and an instance method to do the reverse
*/
public final class TaggedMessage
{
/**
* This message's tag
*/
private ulong tag;

/**
* The payload
*/
private byte[] data;

/**
* Constructs a new TaggedMessage with the given tag and payload
*
* Params:
* tag = the tag to use
* data = the payload
*/
this(ulong tag, byte[] data)
{
this.tag = tag;
this.data = data;
}

/**
* Parameterless constructor used for decoder
*/
private this() {}

/**
* Decodes the wire-formatted tristanable bytes into an instance
* of TaggedMessage whereby the tag and data can be seperately
* accessed and manipulated
*
* Params:
* encodedMessage = the wire-format encoded bytes
* Returns: an instance of TaggedMessage
*/
public static TaggedMessage decode(byte[] encodedMessage)
{
/* The decoded message */
TaggedMessage decodedMessage = new TaggedMessage();

/* The decoded tag */
ulong decodedTag;

/* Take ulong-many bytes and only flip them to LE if not on LE host */
decodedTag = order(bytesToIntegral!(ushort)(cast(ubyte[])encodedMessage), Order.LE);


/* Set the tag */
decodedMessage.setTag(decodedTag);

/* Set the data *(9-th byte onwards) */
decodedMessage.setPayload(encodedMessage[8..$]);

return decodedMessage;
}

/**
* Encodes the tagged message into the tristanable
* wire format ready for transmission
*
* Returns: the encoded bytes
*/
public byte[] encode()
{
/* The encoded bytes */
byte[] encodedMessage;

/* If on little endian then no re-order, if host is BE flip (the tag) */
encodedMessage ~= toBytes(order(tag, Order.LE));


/* Tack on the data */
encodedMessage ~= data;

return encodedMessage;
}

/**
* Get the message's payload
*
* Returns: the payload
*/
public byte[] getPayload()
{
return data;
}

/**
* Get the message's tag
*
* Returns: the tag
*/
public ulong getTag()
{
return tag;
}

/**
* Set the message's payload
*
* Params:
* newPayload = the payload to use
*/
public void setPayload(byte[] newPayload)
{
this.data = newPayload;
}

/**
* Set the message's tag
*
* Params:
* newTag = the tag to use
*/
public void setTag(ulong newTag)
{
this.tag = newTag;
}

/**
* Returns a string representation of the TaggedMessage
*
* Returns: the string represenation
*/
public override string toString()
{
return "TMessage [Tag: "~to!(string)(tag)~", Payload: "~to!(string)(data)~"]";
}
}

/**
* Test encoding and decoding
*/
unittest
{
/* Setup testing data */
TaggedMessage testData = new TaggedMessage(420, [1,2,3]);

/* Encode */
byte[] encoded = testData.encode();

/* Decode */
TaggedMessage decoded = TaggedMessage.decode(encoded);

/* Now ensure that `decoded` == original `testData` */
assert(decoded.getTag() == testData.getTag);
assert(decoded.getPayload() == testData.getPayload());
}
73 changes: 73 additions & 0 deletions source/tristanable/exceptions.d
@@ -0,0 +1,73 @@
/**
* Error handling type definitions
*/
module tristanable.exceptions;

import std.conv : to;

/**
* The type of sub-error of the `TristanableException`
*/
public enum ErrorType
{
/**
* If the requested queue could not be found
*/
QUEUE_NOT_FOUND,

/**
* If the queue wanting to be registered has already
* been registered under the same tag
*/
QUEUE_ALREADY_EXISTS,

/**
* If no default queue is configured
*/
NO_DEFAULT_QUEUE,

/**
* The blocking call to `dequeue()`, somehow, failed
*/
DEQUEUE_FAILED,

/**
* The call to `enqueue()`, somehow, failed
*/
ENQUEUE_FAILED
}

/**
* Any sort of error that occurs during runtime of the tristanable
* engine
*/
public class TristanableException : Exception
{
/**
* The sub-error type
*/
private ErrorType err;

/**
* Constructs a new `TristanableException` with the provided
* sub-error type
*
* Params:
* err = the `ErrorType`
*/
this(ErrorType err)
{
super(this.classinfo.name~": "~to!(string)(err));
this.err = err;
}

/**
* Retrieve the sub-error type
*
* Returns: the sub-error type as a `ErrorType`
*/
public ErrorType getError()
{
return err;
}
}
15 changes: 0 additions & 15 deletions source/tristanable/manager.d

This file was deleted.

34 changes: 34 additions & 0 deletions source/tristanable/manager/config.d
@@ -0,0 +1,34 @@
/**
* Configuration for the manager
*/
module tristanable.manager.config;

/**
* Manager parameters
*/
public struct Config
{
/**
* If set to true then when one uses
* `sendMessage(TaggedMessage)` the
* manager will check if a queue with
* said tag has not been registered
* and if so, then register it for
* us before encoding-and-sending
*/
public bool registerOnSend;
}

/**
* Generates the default configuration to use for
* the manager
*
* Returns: the Config
*/
public Config defaultConfig()
{
Config config;
config.registerOnSend = false;

return config;
}

0 comments on commit b24c336

Please sign in to comment.