Dub

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

.gitignore

@@ -24,3 +24,4 @@ docs/
 *.lst
 source/tristanable/queue.d
 dub.selections.json
+tristanable-test-library

README.md

@@ -1,23 +1,23 @@
-![](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)
 
-**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;
@@ -41,7 +41,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;
@@ -57,7 +57,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
 
@@ -66,8 +66,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

dub.json

@@ -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",
+		"libsnooze": ">=1.3.0-beta"
 	},
 	"description": "Tristanable network message queuing framework",
+	"homepage": "https://deavmi.assigned.network/projects/tristanable",
 	"license": "LGPL-3.0",
 	"name": "tristanable",
 	"targetType": "library"

source/tristanable/encoding.d

@@ -0,0 +1,221 @@
+/**
+ * Encoding/decoding of the tristanable format
+ */
+module tristanable.encoding;
+
+import std.conv : to;
+
+/** 
+ * 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;
+        
+        /* If on little endian then dump direct */
+        version(LittleEndian)
+        {
+            decodedTag = *cast(ulong*)encodedMessage.ptr;
+        }
+        /* If on big endian then reverse received 8 bytes */
+        else version(BigEndian)
+        {
+            /* Base of our tag */
+            byte* tagHighPtr = cast(byte*)decodedTag.ptr;
+
+            *(tagHighPtr+0) = encodedMessage[7];
+            *(tagHighPtr+1) = encodedMessage[6];
+            *(tagHighPtr+2) = encodedMessage[5];
+            *(tagHighPtr+3) = encodedMessage[4];
+            *(tagHighPtr+4) = encodedMessage[3];
+            *(tagHighPtr+5) = encodedMessage[2];
+            *(tagHighPtr+6) = encodedMessage[1];
+            *(tagHighPtr+7) = encodedMessage[0];
+        }
+        /* Blessed is the fruit of thy womb Jesus, hail Mary, mother of God, pray for our sinners - now and at the hour of our death - Amen */
+        else
+        {
+            pragma(msg, "Not too sure about tha 'ey 😳️");
+        }
+
+        /* 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 dump 64 bit as is - little endian */
+        version(LittleEndian)
+        {
+            /* Base (little first) of tag */
+            byte* basePtr = cast(byte*)&tag;
+
+            encodedMessage ~= *(basePtr+0);
+            encodedMessage ~= *(basePtr+1);
+            encodedMessage ~= *(basePtr+2);
+            encodedMessage ~= *(basePtr+3);
+            encodedMessage ~= *(basePtr+4);
+            encodedMessage ~= *(basePtr+5);
+            encodedMessage ~= *(basePtr+6);
+            encodedMessage ~= *(basePtr+7);
+        }
+        /* If on big endian, then traverse 64-bit number in reverse - and tack on */
+        else version(BigEndian)
+        {
+            /* Base (biggest first) of tag */
+            byte* highPtr = cast(byte*)&tag;
+
+            encodedMessage ~= *(highPtr+7);
+            encodedMessage ~= *(highPtr+6);
+            encodedMessage ~= *(highPtr+5);
+            encodedMessage ~= *(highPtr+4);
+            encodedMessage ~= *(highPtr+3);
+            encodedMessage ~= *(highPtr+2);
+            encodedMessage ~= *(highPtr+1);
+            encodedMessage ~= *(highPtr+0);
+        }
+        /* Hail marry, mother of God, pray for our sinners, now and at the our of our death Amen */
+        else
+        {
+            pragma(msg, "Not feeling scrumptious homeslice 😎️");
+        }
+
+        /* 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());
+}

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;
+    }
+}

source/tristanable/manager.d

@@ -1,15 +0,0 @@
-module tristanable.manager;
-
-import tristanable.queue : Queue;
-
-/** 
- * Allows one to add new queues, control
- * existing ones by waiting on them etc
- */
-public class Manager
-{
-    /* Queues */
-    private Queue[] queues;
-
-    
-}

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;
+}

source/tristanable/manager/manager.d

@@ -0,0 +1,549 @@
+/** 
+ * Management of a tristanable instance
+ */
+module tristanable.manager.manager;
+
+import std.socket;
+import tristanable.queue : Queue;
+import core.sync.mutex : Mutex;
+import tristanable.manager.watcher : Watcher;
+import tristanable.encoding : TaggedMessage;
+import tristanable.exceptions;
+import std.container.slist : SList;
+import tristanable.manager.config;
+import river.core;
+import river.impls.sock : SockStream;
+import bformat.client;
+
+/** 
+ * Manages a provided socket by spawning
+ * a watcher thread to read from it and file
+ * mail into the corresponding queues.
+ *
+ * Queues are managed via this an instance
+ * of a manager.
+ */
+public class Manager
+{
+    /** 
+     * Configuration
+     */
+    private Config config;
+
+    /** 
+     * The bformat client to read and write from
+     */
+    private BClient bClient;
+
+    /** 
+     * Currently registered queues
+     */
+    private SList!(Queue) queues;
+
+    /** 
+     * Lock for currently registered queues
+     */
+    private Mutex queuesLock;
+
+    /** 
+     * Default queue
+     */
+    private Queue defaultQueue;
+
+    /** 
+     * Watcher which manages the socket and
+     * enqueues new messages into the respective
+     * quueue for us
+     */
+    private Watcher watcher;
+
+    /** 
+     * Constructs a new manager which will read from
+     * this socket and file mail for us
+     *
+     * Params:
+     *   stream = the underlying stream to use
+     */
+    this(Stream stream, Config config = defaultConfig())
+    {
+        this.bClient = new BClient(stream);
+        this.queuesLock = new Mutex();
+        this.config = config;
+        this.watcher = new Watcher(this, bClient);
+    }
+
+    // TODO: Comment this
+    // This is for backwards compatibility (whereby a `Socket` was taken in)
+    this(Socket socket, Config config = defaultConfig())
+    {
+        this(new SockStream(socket), config);
+    }
+
+    /** 
+     * Starts the management of the socket,
+     * resulting in queues being updated upon
+     * reciving messages tagged for them
+     */
+    public void start()
+    {
+        watcher.startWatcher();
+    }
+
+    /** 
+     * Stops the management of the socket, resulting
+     * in ending the updating of queues and closing
+     * the underlying connection
+     */
+    public void stop()
+    {
+        watcher.shutdown();
+    }
+
+    /**
+     * Retrieves the queue mathcing the provided id
+     *
+     * Params:
+     *   id = the id to lookup by
+     * Returns: the Queue
+     * Throws: TristanableException if the queue is not found
+     */
+    public Queue getQueue(ulong id)
+    {
+        /* The found queue */
+        Queue queue = getQueue_nothrow(id);
+
+        /* If no queue is found then throw an error */
+        if(queue is null)
+        {
+            throw new TristanableException(ErrorType.QUEUE_NOT_FOUND);
+        }
+
+        return queue;
+    }
+
+    /** 
+     * Retrieves the queue mathcing the provided id
+     *
+     * This is the nothrow version
+     *
+     * Params:
+     *   id = the id to lookup by
+     * Returns: the Queue if found, null otherwise
+     */
+    public Queue getQueue_nothrow(ulong id)
+    {
+        /* The found queue */
+        Queue queue;
+
+        /* Lock the queue of queues */
+        queuesLock.lock();
+
+        /* On return or error */
+        scope(exit)
+        {
+            /* Unlock the queue of queues */
+            queuesLock.unlock();
+        }
+
+        /* Search for the queue */
+        foreach(Queue curQueue; queues)
+        {
+            if(curQueue.getID() == id)
+            {
+                queue = curQueue;
+                break;
+            }
+        }
+
+        return queue;
+    }
+
+    /** 
+     * Get a new queue thatis unique in its tag
+     * (unused/not regustered yet), register it
+     * and then return it
+     *
+     * Returns: the newly registered Queue
+     */
+    public Queue getUniqueQueue()
+    {
+        /* The newly created queue */
+        Queue uniqueQueue;
+
+        /* Lock the queue of queues */
+        queuesLock.lock();
+
+        /* On return or error */
+        scope(exit)
+        {
+            /* Unlock the queue of queues */
+            queuesLock.unlock();
+        }
+
+        // TODO: Throw exception if all tags used
+        /* The unused tag */
+        ulong unusedTag = 0;
+
+        /* Try the current tag and ensure no queue uses it */
+        tagLoop: for(ulong curPotentialTag = 0; true; curPotentialTag++)
+        {
+            foreach(Queue curQueue; queues)
+            {
+                if(curQueue.getID() == curPotentialTag)
+                {
+                    continue tagLoop;
+                }
+            }
+
+            /* Then we have found a unique tag */
+            unusedTag = curPotentialTag;
+            break;
+        }
+
+        /* Create the queue */
+        uniqueQueue = new Queue(unusedTag);
+
+        /* Register it */
+        registerQueue(uniqueQueue);
+
+        return uniqueQueue;
+    }
+
+    /** 
+     * Registers the given queue with the manager
+     *
+     * Params:
+     *   queue = the queue to register
+     * Throws:
+     *   TristanableException if a queue with the provided id already exists
+     */
+    public void registerQueue(Queue queue)
+    {
+        /* Try to register the queue */
+        bool status = registerQueue_nothrow(queue);
+
+        /* If registration was not successful */
+        if(!status)
+        {
+            throw new TristanableException(ErrorType.QUEUE_ALREADY_EXISTS);
+        }
+    }
+
+    /** 
+     * Registers the given queue with the manager
+     *
+     * Params:
+     *   queue = the queue to register
+     * Returns: true if registration was successful, false otherwise
+     */
+    public bool registerQueue_nothrow(Queue queue)
+    {
+        /* Lock the queue of queues */
+        queuesLock.lock();
+
+        /* On return or error */
+        scope(exit)
+        {
+            /* Unlock the queue of queues */
+            queuesLock.unlock();
+        }
+
+        /* Search for the queue, throw an exception if it exists */
+        foreach(Queue curQueue; queues)
+        {
+            if(curQueue.getID() == queue.getID())
+            {
+                /* Registration failed */
+                return false;
+            }
+        }
+
+        /* Insert the queue as it does not exist */
+        queues.insertAfter(queues[], queue);
+
+        /* Registration was a success */
+        return true;
+    }
+
+    /** 
+     * De-registers the given queue from the manager
+     *
+     * Params:
+     *   queue = the queue to de-register
+     * Throws:
+     *   TristanableException if a queue with the provided id cannot be found
+     */
+    public void releaseQueue(Queue queue)
+    {
+        /* Try to de-register the queue */
+        bool status = releaseQueue_nothrow(queue);
+
+        /* If de-registration was not successful */
+        if(!status)
+        {
+            throw new TristanableException(ErrorType.QUEUE_NOT_FOUND);
+        }
+    }
+
+    /** 
+     * De-registers the given queue from the manager
+     *
+     * Params:
+     *   queue = the queue to de-register
+     * Returns: true if de-registration was successful, false otherwise
+     */
+    public bool releaseQueue_nothrow(Queue queue)
+    {
+        /* Lock the queue of queues */
+        queuesLock.lock();
+
+        /* On return or error */
+        scope(exit)
+        {
+            /* Unlock the queue of queues */
+            queuesLock.unlock();
+        }
+
+        /* Search for the queue, return false if it does NOT exist */
+        foreach(Queue curQueue; queues)
+        {
+            if(curQueue.getID() == queue.getID())
+            {
+                /* Remove the queue */
+                queues.linearRemoveElement(queue);
+
+                /* De-registration succeeded */
+                return true;
+            }
+        }
+
+        /* De-registration failed */
+        return false;
+    }
+
+    /** 
+     * Sets the default queue
+     *
+     * The default queue, when set/enabled, is the queue that will
+     * be used to enqueue messages that have a tag which doesn't
+     * match any of the normally registered queues.
+     *
+     * Please note that the ID of the queue passed in here does not
+     * mean anything in this context; only the queuing facilities
+     * of the Queue object are used
+     *
+     * Params:
+     *   queue = the default queue to use
+     */
+    public void setDefaultQueue(Queue queue)
+    {
+        this.defaultQueue = queue;
+    }
+
+    /** 
+     * Returns the default queue
+     *
+     * Returns: the default queue
+     * Throws:
+     *   TristanableException if there is no default queue
+     */
+    public Queue getDefaultQueue()
+    {
+        /* The potential default queue */
+        Queue potentialDefaultQueue = getDefaultQueue_nothrow();
+
+        if(potentialDefaultQueue is null)
+        {
+            throw new TristanableException(ErrorType.NO_DEFAULT_QUEUE);
+        }
+
+        return potentialDefaultQueue;
+    }
+
+    /** 
+     * Returns the default queue
+     *
+     * This is the nothrow version
+     *
+     * Returns: the default queue if found, null otherwise
+     */
+    public Queue getDefaultQueue_nothrow()
+    {
+        return defaultQueue;
+    }
+
+    /** 
+     * Sends the provided message over the socket
+     *
+     * Params:
+     *   message = the TaggedMessage to send
+     */
+    public void sendMessage(TaggedMessage message)
+    {
+        /**
+         * If a queue with the tag of the message does
+         * not exist, then register it if the config
+         * option was enabled
+         */
+        if(config.registerOnSend)
+        {
+            /* Create a Queue with the tag */
+            Queue createdQueue = new Queue(message.getTag());
+
+            /* Attempt to register the queue */
+            registerQueue_nothrow(createdQueue);
+        }
+
+        /* Encode the message */
+        byte[] encodedMessage = message.encode();
+
+        /* Send it using bformat (encode-and-send) */
+        bClient.sendMessage(encodedMessage);
+    }
+}
+
+
+
+// TODO: Fix this, write it in a nicer way
+// ... or make a private constructor here that
+// ... does not take it in
+version(unittest)
+{
+    Socket nullSock = null;
+}
+
+/**
+ * Test retrieving a queue which does not
+ * exist
+ */
+unittest
+{
+    /* Create a manager */
+    Manager manager = new Manager(nullSock);
+
+    /* Shouldn't be found */
+    try
+    {
+        manager.getQueue(69);
+        assert(false);
+    }
+    catch(TristanableException e)
+    {
+        assert(e.getError() == ErrorType.QUEUE_NOT_FOUND);
+    }
+
+    /* Shouldn't be found */
+    assert(manager.getQueue_nothrow(69) is null);
+}
+
+/**
+ * Test registering a queue and then fetching it
+ */
+unittest
+{
+    /* Create a manager */
+    Manager manager = new Manager(nullSock);
+
+    /* Create a new queue with tag 69 */
+    Queue queue = new Queue(69);
+
+    try
+    {
+        /* Register the queue */
+        manager.registerQueue(queue);
+
+        /* Fetch the queue */
+        Queue fetchedQueue = manager.getQueue(69);
+
+        /* Ensure the queue we fetched is the one we stored (the references would be equal) */
+        assert(fetchedQueue == queue);
+    }
+    catch(TristanableException e)
+    {
+        assert(false);
+    }
+
+    /* Should be found */
+    assert(manager.getQueue_nothrow(69) !is null);
+}
+
+/**
+ * Tests registering a queue and then registering
+ * another queue with the same id
+ */
+unittest
+{
+    /* Create a manager */
+    Manager manager = new Manager(nullSock);
+
+    /* Create a new queue with tag 69 */
+    Queue queue = new Queue(69);
+
+    /* Register the queue */
+    manager.registerQueue(queue);
+
+    try
+    {
+        /* Register the queue (try again) */
+        manager.registerQueue(queue);
+
+        assert(false);
+    }
+    catch(TristanableException e)
+    {
+        assert(e.getError() == ErrorType.QUEUE_ALREADY_EXISTS);
+    }
+}
+
+/**
+ * Tests registering a queue, de-registering it and
+ * then registering it again
+ */
+unittest
+{
+    /* Create a manager */
+    Manager manager = new Manager(nullSock);
+
+    /* Create a new queue with tag 69 */
+    Queue queue = new Queue(69);
+
+    /* Register the queue */
+    manager.registerQueue(queue);
+
+    /* Ensure it is registered */
+    assert(queue == manager.getQueue(69));
+
+    /* De-register the queue */
+    manager.releaseQueue(queue);
+
+    /* Ensure it is de-registered */
+    assert(manager.getQueue_nothrow(69) is null);
+
+    /* Register the queue (again) */
+    manager.registerQueue(queue);
+
+    /* Ensure it is registered (again) */
+    assert(queue == manager.getQueue(69));
+}
+
+/**
+ * Tests registering a queue using the "next available queue"
+ * method
+ */
+unittest
+{
+    /* Create a manager */
+    Manager manager = new Manager(nullSock);
+
+    /* Get the next 3 available queues */
+    Queue queue1 = manager.getUniqueQueue();
+    Queue queue2 = manager.getUniqueQueue();
+    Queue queue3 = manager.getUniqueQueue();
+
+    /* The queues should have tags [0, 1, 2] respectively */
+    assert(queue1.getID() == 0);
+    assert(queue2.getID() == 1);
+    assert(queue3.getID() == 2);
+}
+
+// TODO: Add testing for queue existence (internal method)

source/tristanable/manager/package.d

@@ -0,0 +1,15 @@
+/**
+ * Interface which manages a provided socket
+ * and enqueuing and dequeuing of queues
+ */
+module tristanable.manager;
+
+/**
+ * The management facilities
+ */
+public import tristanable.manager.manager : Manager;
+
+/** 
+ * Configuration for the manager
+ */
+public import tristanable.manager.config : Config, defaultConfig;

source/tristanable/manager/watcher.d

@@ -0,0 +1,258 @@
+/**
+ * Facilitates the reading of messages from the socket,
+ * decoding thereof and final enqueuing thereof into their
+ * respective queus
+ */
+module tristanable.manager.watcher;
+
+import core.thread : Thread;
+import tristanable.manager.manager : Manager;
+import std.socket;
+import bformat;
+import tristanable.encoding;
+import tristanable.exceptions;
+import tristanable.queue;
+import bformat.client;
+
+/** 
+ * Watches the socket on a thread of its own,
+ * performs the decoding of the incoming messages
+ * and places them into the correct queues via
+ * the associated Manager instance
+ */
+public class Watcher : Thread
+{
+    /** 
+     * The associated manager to use
+     * such that we can place new mail
+     * into their respective inboxes (queues)
+     */
+    private Manager manager;
+
+    /** 
+     * The BClient to read from
+     */
+    private BClient bClient;
+
+    /** 
+     * Creates a new `Watcher` that is associated
+     * with the provided `Manager` such that it can
+     * add to its registered queues. The provided `Socket`
+     * is such that it can be read from and managed.
+     *
+     * Params:
+     *   manager = the `Manager` to associate with
+     *   bclient = the underlying `BClient` to read data from
+     */
+    package this(Manager manager, BClient bClient)
+    {
+        this.manager = manager;
+        this.bClient = bClient;
+
+        super(&watch);
+    }
+
+    /** 
+     * Starts the underlying thread
+     */
+    package void startWatcher()
+    {
+        /* Start the watch method on a new thread */
+        start();
+    }
+
+    /** 
+     * Watches the socket for incoming messages
+     * and decodes them on the fly, placing
+     * the final message in the respective queue
+     */    
+    private void watch()
+    {
+        import std.stdio;
+        
+        while(true)
+        {
+            /* Do a bformat read-and-decode */
+            byte[] wireTristan;
+            version(unittest) { writeln("Before bformat recv()"); }
+            bool recvStatus = bClient.receiveMessage(wireTristan); // TODO: Add a check for the status of read
+            version(unittest) { writeln("After bformat recv()"); }
+            version(unittest) { writeln("bformat recv() status: ", recvStatus); }
+
+            if(recvStatus)
+            {
+                /* Decode the received bytes into a tagged message */
+                TaggedMessage decodedMessage = TaggedMessage.decode(wireTristan);
+                version(unittest) { writeln("Watcher received: ", decodedMessage); }
+
+                /* Search for the queue with the id provided */
+                ulong messageTag = decodedMessage.getTag();
+                Queue potentialQueue = manager.getQueue_nothrow(messageTag);
+
+                /* If a queue can be found */
+                if(potentialQueue !is null)
+                {
+                    /* Enqueue the message */
+                    potentialQueue.enqueue(decodedMessage);
+                }
+                /* If the queue if not found */
+                else
+                {
+                    /**
+                     * Look for a default queue, and if one is found
+                     * then enqueue the message there. Otherwise, drop
+                     * it by simply doing nothing.
+                     */
+                    try
+                    {
+                        potentialQueue = manager.getDefaultQueue();
+
+                        /* Enqueue the message */
+                        potentialQueue.enqueue(decodedMessage);
+                    }
+                    catch(TristanableException e) {}
+                }
+
+                version(unittest) { writeln("drip"); }
+            }
+            /**
+             * If there was an error receiving on the socket.
+             *
+             * This can be either because we have shut the socket down
+             * or the remote end has closed the connection.
+             *
+             * In any case, exit the loop therefore ending this thread.
+             */
+            else
+            {
+                break;
+            }
+        }
+    }
+
+    /** 
+     * Shuts down the watcher, unblocks the blocking read in the loop
+     * resulting in the watcher thread ending
+     */
+    package void shutdown()
+    {
+        /* Closes the bformat reader */
+        bClient.close();
+    }
+}
+
+/** 
+ * Set up a server which will send some tagged messages to us (the client),
+ * where we have setup a `Manager` to watch the queues with tags `42` and `69`,
+ * we then dequeue some messages from both queus. Finally, we shut down the manager.
+ */
+unittest
+{
+    import std.socket;
+    import std.stdio;
+    import core.thread;
+
+    Address serverAddress = parseAddress("::1", 0);
+    Socket server = new Socket(AddressFamily.INET6, SocketType.STREAM, ProtocolType.TCP);
+    server.bind(serverAddress);
+    server.listen(0);
+
+    class ServerThread : Thread
+    {
+        this()
+        {
+            super(&worker);
+        }
+
+        private void worker()
+        {
+            Socket clientSocket = server.accept();
+            BClient bClient = new BClient(clientSocket);
+
+            Thread.sleep(dur!("seconds")(7));
+            writeln("Server start");
+
+            /** 
+             * Create a tagged message to send
+             *
+             * tag 42 payload Cucumber 😳️
+             */
+            TaggedMessage message = new TaggedMessage(42, cast(byte[])"Cucumber 😳️");
+            byte[] tEncoded = message.encode();
+            writeln("server send status: ", bClient.sendMessage(tEncoded));
+
+            writeln("server send [done]");
+
+            /** 
+             * Create a tagged message to send
+             *
+             * tag 69 payload Hello
+             */
+            message = new TaggedMessage(69, cast(byte[])"Hello");
+            tEncoded = message.encode();
+            writeln("server send status: ", bClient.sendMessage(tEncoded));
+
+            writeln("server send [done]");
+
+            /** 
+             * Create a tagged message to send
+             *
+             * tag 69 payload Bye
+             */
+            message = new TaggedMessage(69, cast(byte[])"Bye");
+            tEncoded = message.encode();
+            writeln("server send status: ", bClient.sendMessage(tEncoded));
+
+            writeln("server send [done]");
+        }
+    }
+
+    ServerThread serverThread = new ServerThread();
+    serverThread.start();
+
+    Socket client = new Socket(AddressFamily.INET6, SocketType.STREAM, ProtocolType.TCP);
+    
+    writeln(server.localAddress);
+
+
+    Manager manager = new Manager(client);
+
+    Queue sixtyNine = new Queue(69);
+    Queue fortyTwo = new Queue(42);
+
+    manager.registerQueue(sixtyNine);
+    manager.registerQueue(fortyTwo);
+
+
+    /* Connect our socket to the server */
+    client.connect(server.localAddress);
+
+    /* Start the manager and let it manage the socket */
+    manager.start();
+
+    /* Block on the unittest thread for a received message */
+    writeln("unittest thread: Dequeue() blocking...");
+    TaggedMessage dequeuedMessage = sixtyNine.dequeue();
+    writeln("unittest thread: Got '"~dequeuedMessage.toString()~"' decode payload to string '"~cast(string)dequeuedMessage.getPayload()~"'");
+    assert(dequeuedMessage.getTag() == 69);
+    assert(dequeuedMessage.getPayload() == cast(byte[])"Hello");
+
+    /* Block on the unittest thread for a received message */
+    writeln("unittest thread: Dequeue() blocking...");
+    dequeuedMessage = sixtyNine.dequeue();
+    writeln("unittest thread: Got '"~dequeuedMessage.toString()~"' decode payload to string '"~cast(string)dequeuedMessage.getPayload()~"'");
+    assert(dequeuedMessage.getTag() == 69);
+    assert(dequeuedMessage.getPayload() == cast(byte[])"Bye");
+
+    /* Block on the unittest thread for a received message */
+    writeln("unittest thread: Dequeue() blocking...");
+    dequeuedMessage = fortyTwo.dequeue();
+    writeln("unittest thread: Got '"~dequeuedMessage.toString()~"' decode payload to string '"~cast(string)dequeuedMessage.getPayload()~"'");
+    assert(dequeuedMessage.getTag() == 42);
+    assert(dequeuedMessage.getPayload() == cast(byte[])"Cucumber 😳️");
+    
+    
+
+    /* Stop the manager */
+    manager.stop();
+}

source/tristanable/package.d

@@ -3,6 +3,23 @@
  */
 module tristanable;
 
+/**
+ * Interface which manages a provided socket
+ * and enqueuing and dequeuing of queues
+ */
+public import tristanable.manager;
 
-public import tristanable.manager : Manager;
-public import tristanable.queue : Queue, QueueItem;
+/**
+ * A queue of queue items all of the same tag
+ */
+public import tristanable.queue : Queue;
+
+/**
+ * Error handling type definitions
+ */
+public import tristanable.exceptions : TristanableException, ErrorType;
+
+/**
+ * Encoding/decoding of the tristanable format
+ */
+public import tristanable.encoding : TaggedMessage;

source/tristanable/queue.d

@@ -1,53 +1,223 @@
+/**
+ * A queue of queue items all of the same tag
+ */
 module tristanable.queue;
 
+// TODO: Examine the below import which seemingly fixes stuff for libsnooze
+import libsnooze.clib;
 import libsnooze;
-import core.sync.mutex : Mutex;
 
+import core.sync.mutex : Mutex;
+import std.container.slist : SList;
+import tristanable.encoding;
+import core.thread : dur;
+import tristanable.exceptions;
+
+version(unittest)
+{
+   import std.stdio;
+   import std.conv : to;
+}
+
+/** 
+ * Represents a queue whereby messages of a certain tag/id
+ * can be enqueued to (by the `Watcher`) and dequeued from
+ * (by the user application)
+ */
 public class Queue
 {
     /** 
-     * Everytime a thread calls `.dequeue()` on this queue
-     * 
+     * The libsnooze event used to sleep/wake
+     * on queue events
      */
     private Event event;
 
-    private QueueItem queue;
+    /** 
+     * The queue of messages
+     */
+    private SList!(TaggedMessage) queue;
+
+    /** 
+     * The lock for the message queue
+     */
     private Mutex queueLock;
+    
+    /** 
+     * This queue's unique ID
+     */
+    private ulong queueID;
 
 
-    private this()
+    /** 
+     * Constructs a new Queue and immediately sets up the notification
+     * sub-system for the calling thread (the thread constructing this
+     * object) which ensures that a call to dequeue will immediately
+     * unblock on the first message received under this tag
+     *
+     * Params:
+     *   queueID = the id to use for this queue
+     */
+    this(ulong queueID)
     {
         /* Initialize the queue lock */
         this.queueLock = new Mutex();
 
         /* Initialize the event */
         this.event = new Event();
+
+        /* Set the queue id */
+        this.queueID = queueID;
+
+        /* Ensure pipe existence (see https://deavmi.assigned.network/git/deavmi/tristanable/issues/5) */
+        event.wait(dur!("seconds")(0));
     }
 
-    public void dequeue()
+    /** 
+     * Enqueues the provided tagged message onto this queue
+     * and then wakes up any thread that has called dequeue
+     * on this queue as well
+     *
+     * On error enqueueing a `TristanableException` will be
+     * thrown.
+     *
+     * Params:
+     *   message = the TaggedMessage to enqueue
+     */
+    public void enqueue(TaggedMessage message)
     {
-        // TODO: Make us wait on the event (optional with a time-out)
+        version(unittest)
+        {
+            writeln("queue["~to!(string)(queueID)~"]: Enqueuing '"~to!(string)(message)~"'...");
+        }
 
-        // TODO: Lock queue
+        scope(exit)
+        {
+            version(unittest)
+            {
+                writeln("queue["~to!(string)(queueID)~"]: Enqueued '"~to!(string)(message)~"'!");
+            }
+
+            /* Unlock the item queue */
+            queueLock.unlock();
+        }
+
+        /* Lock the item queue */
         queueLock.lock();
 
-        // TODO: Get item off queue
+        /* Add the item to the queue */
+        queue.insertAfter(queue[], message);
 
-        // TODO: Unlock queue
-        queueLock.unlock();
+        /* Wake up anyone wanting to dequeue from us */
+        try
+        {
+            // TODO: Make us wait on the event (optional with a time-out)
+            event.notifyAll();
+        }
+        catch(FatalException snozErr)
+        {
+            // Throw an exception on a fatal exception
+            throw new TristanableException(ErrorType.ENQUEUE_FAILED);
+        }
     }
 
-    public static Queue newQueue(ulong queueID)
+    // TODO: Make a version of this which can time out
+
+    /** 
+     * Blocks till a message can be dequeued from this queue
+     *
+     * On error dequeueing a `TristanableException` will be
+     * thrown.
+     *
+     * Returns: the dequeued TaggedMessage
+     */
+    public TaggedMessage dequeue()
     {
-        Queue queue;
+        version(unittest)
+        {
+            writeln("queue["~to!(string)(queueID)~"]: Dequeueing...");
+        }
 
-        // TODO: Implement me
+        /* The dequeued message */
+        TaggedMessage dequeuedMessage;
 
-        return queue;
+        scope(exit)
+        {
+            version(unittest)
+            {
+                writeln("queue["~to!(string)(queueID)~"]: Dequeued '"~to!(string)(dequeuedMessage)~"'!");
+            }
+        }
+
+        /* Block till we dequeue a message successfully */
+        while(dequeuedMessage is null)
+        {
+            /**
+             * Call `wait()` and catch any interrupts
+             * in which case loop back and call `wait()`
+             * again
+             */
+            while(true)
+            {
+                try
+                {
+                    // TODO: Make us wait on the event (optional with a time-out)
+                    event.wait();
+                }
+                catch(InterruptedException e)
+                {
+                    version(unittest)
+                    {
+                        import std.stdio;
+                        writeln("dequeue() had libsnooze wait() get interrupted!");
+                    }
+
+                    // Retry the wait()
+                    continue;
+                }
+                catch(FatalException fatalErr)
+                {
+                    version(unittest)
+                    {
+                        import std.stdio;
+                        writeln("dequeue() had libsnooze wait() get FATALLY fail! Exception will now throw...");
+                    }
+
+                    // Throw an exception on a fatal exception
+                    throw new TristanableException(ErrorType.DEQUEUE_FAILED);
+                }
+
+                // On successful wait() wake-up exit this wait()-retry loop
+                break;
+            }
+            
+
+            /* Lock the item queue */
+            queueLock.lock();
+
+            /* Consume the front of the queue (if non-empty) */
+            if(!queue.empty())
+            {
+                /* Pop the front item off */
+                dequeuedMessage = queue.front();
+
+                /* Remove the front item from the queue */
+                queue.linearRemoveElement(dequeuedMessage);
+            }
+
+            /* Unlock the item queue */
+            queueLock.unlock();
+        }
+
+        return dequeuedMessage;
     }
-}
 
-public class QueueItem
-{
-    
+    /** 
+     * Get the id/tag of this queue
+     *
+     * Returns: the queue's id
+     */
+    public ulong getID()
+    {
+        return queueID;
+    }
 }