.github/workflows/d.yml

@@ -6,9 +6,9 @@ name: D
 
 on:
   push:
-    branches: [ "**" ]
+    branches: [ "master" ]
   pull_request:
-    branches: [ "**" ]
+    branches: [ "master", "nextgen" ]
 
 permissions:
   contents: read
@@ -22,11 +22,6 @@ jobs:
     - uses: actions/checkout@v3
     - uses: dlang-community/setup-dlang@4c99aa991ce7d19dd3064de0a4f2f6b2f152e2d7
 
-    - name: Install Doveralls (code coverage tool)
-      run: |
-        dub fetch doveralls
-        sudo apt install libcurl4-openssl-dev
-
     - name: 'Build & Test'
       run: |
         # Build the project, with its main file included, without unittests
@@ -34,8 +29,4 @@ jobs:
         # Build and run tests, as defined by `unittest` configuration
         # In this mode, `mainSourceFile` is excluded and `version (unittest)` are included
         # See https://dub.pm/package-format-json.html#configurations
-        dub test --compiler=$DC --coverage
-
-    - name: Coverage upload
-      run: |
-        dub run doveralls -- -t ${{secrets.COVERALLS_REPO_TOKEN}}
+        dub test --compiler=$DC

README.md

@@ -3,8 +3,7 @@
 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)
-
+[![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 respectively tagged queues indicated by their _"type"_ or `id`.
 

dub.json

@@ -4,8 +4,8 @@
 	],
 	"copyright": "Copyright © 2023, Tristan B. Kildaire",
 	"dependencies": {
-		"bformat": ">=4.1.1",
-		"niknaks": ">=0.3.0"
+		"bformat": "4.1.0",
+		"libsnooze": "0.3.3"
 	},
 	"description": "Tristanable network message queuing framework",
 	"homepage": "https://deavmi.assigned.network/projects/tristanable",

source/tristanable/encoding.d

@@ -4,7 +4,6 @@
 module tristanable.encoding;
 
 import std.conv : to;
-import niknaks.bits : bytesToIntegral, Order, order, toBytes;
 
 /** 
  * Represents a tagged message that has been decoded
@@ -61,9 +60,31 @@ public final class 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);
-        
+        /* 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);
@@ -85,9 +106,41 @@ public final class TaggedMessage
         /* 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));
+        /* 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;

source/tristanable/exceptions.d

@@ -24,17 +24,7 @@ public enum ErrorType
     /** 
      * 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
+    NO_DEFAULT_QUEUE
 }
 
 /** 

source/tristanable/manager/manager.d

@@ -4,7 +4,7 @@
 module tristanable.manager.manager;
 
 import std.socket;
-import tristanable.queue.queue : Queue;
+import tristanable.queue : Queue;
 import core.sync.mutex : Mutex;
 import tristanable.manager.watcher : Watcher;
 import tristanable.encoding : TaggedMessage;

source/tristanable/manager/watcher.d

@@ -11,7 +11,7 @@ import std.socket;
 import bformat;
 import tristanable.encoding;
 import tristanable.exceptions;
-import tristanable.queue.queue;
+import tristanable.queue;
 import bformat.client;
 
 /** 
@@ -204,28 +204,6 @@ unittest
             writeln("server send status: ", bClient.sendMessage(tEncoded));
 
             writeln("server send [done]");
-
-            /** 
-             * Create a tagged message to send
-             *
-             * tag 100 payload Bye
-             */
-            message = new TaggedMessage(100, cast(byte[])"DEFQUEUE_1");
-            tEncoded = message.encode();
-            writeln("server send status: ", bClient.sendMessage(tEncoded));
-
-            writeln("server send [done]");
-
-            /** 
-             * Create a tagged message to send
-             *
-             * tag 200 payload Bye
-             */
-            message = new TaggedMessage(200, cast(byte[])"DEFQUEUE_2");
-            tEncoded = message.encode();
-            writeln("server send status: ", bClient.sendMessage(tEncoded));
-
-            writeln("server send [done]");
         }
     }
 
@@ -245,10 +223,6 @@ unittest
     manager.registerQueue(sixtyNine);
     manager.registerQueue(fortyTwo);
 
-    // Register a default queue (tag ignored)
-    Queue defaultQueue = new Queue(2332);
-    manager.setDefaultQueue(defaultQueue);
-
 
     /* Connect our socket to the server */
     client.connect(server.localAddress);
@@ -278,19 +252,6 @@ unittest
     assert(dequeuedMessage.getPayload() == cast(byte[])"Cucumber 😳️");
     
     
-    /* Dequeue two messages from the default queue */
-    writeln("unittest thread: Dequeue() blocking...");
-    dequeuedMessage = defaultQueue.dequeue();
-    writeln("unittest thread: Got '"~dequeuedMessage.toString()~"' decode payload to string '"~cast(string)dequeuedMessage.getPayload()~"'");
-    assert(dequeuedMessage.getTag() == 100);
-    assert(dequeuedMessage.getPayload() == cast(byte[])"DEFQUEUE_1");
-
-    writeln("unittest thread: Dequeue() blocking...");
-    dequeuedMessage = defaultQueue.dequeue();
-    writeln("unittest thread: Got '"~dequeuedMessage.toString()~"' decode payload to string '"~cast(string)dequeuedMessage.getPayload()~"'");
-    assert(dequeuedMessage.getTag() == 200);
-    assert(dequeuedMessage.getPayload() == cast(byte[])"DEFQUEUE_2");
-
 
     /* Stop the manager */
     manager.stop();

source/tristanable/package.d

@@ -11,8 +11,9 @@ public import tristanable.manager;
 
 /**
  * A queue of queue items all of the same tag
+ * and queue-related facilities
  */
-public import tristanable.queue.queue : Queue;
+public import tristanable.queue;
 
 /**
  * Error handling type definitions

source/tristanable/queue/queue.d

@@ -5,13 +5,14 @@ module tristanable.queue.queue;
 
 import tristanable.queue.listener : TListener;
 
+// TODO: Examine the below import which seemingly fixes stuff for libsnooze
+import libsnooze.clib;
+import libsnooze;
+
 import core.sync.mutex : Mutex;
-import core.sync.condition : Condition;
-import core.sync.exception : SyncError;
 import std.container.slist : SList;
 import tristanable.encoding;
-import core.time : Duration, dur;
-import tristanable.exceptions;
+import core.thread : dur;
 
 version(unittest)
 {
@@ -34,15 +35,8 @@ public class Queue
     /** 
      * The libsnooze event used to sleep/wake
      * on queue events
-     * Mutex for the condition variable
      */
-    private Mutex mutex;
-
-    /** 
-     * The condition variable used to sleep/wake
-     * on queue of events
-     */
-    private Condition signal;
+    private Event event;
 
     /** 
      * The queue of messages
@@ -66,15 +60,6 @@ public class Queue
 
     // TODO: Add listener add/remove methods
     // TODO: On queue actions add a notificaiton call to the listeners
-    /**
-     * If a message is enqueued prior
-     * to us sleeping then we won't
-     * wake up and return for it.
-     *
-     * Therefore a periodic wakeup
-     * is required.
-     */
-    private Duration wakeInterval;
 
     /** 
      * Constructs a new Queue and immediately sets up the notification
@@ -90,37 +75,14 @@ public class Queue
         /* Initialize the queue lock */
         this.queueLock = new Mutex();
 
-        /* Initialize the condition variable */
-        this.mutex = new Mutex();
-        this.signal = new Condition(this.mutex);
+        /* Initialize the event */
+        this.event = new Event();
 
         /* Set the queue id */
         this.queueID = queueID;
 
-        /* Set the slumber interval */
-        this.wakeInterval = dur!("msecs")(50); // TODO: Decide on value
-    }
-
-    /** 
-     * Returns the current wake interval
-     * for the queue checker
-     *
-     * Returns: the `Duration`
-     */
-    public Duration getWakeInterval()
-    {
-        return this.wakeInterval;
-    }
-
-    /** 
-     * Sets the wake up interval
-     *
-     * Params:
-     *   interval = the new interval
-     */
-    public void setWakeInterval(Duration interval)
-    {
-        this.wakeInterval = interval;
+        /* Ensure pipe existence (see https://deavmi.assigned.network/git/deavmi/tristanable/issues/5) */
+        event.wait(dur!("seconds")(0));
     }
 
     /** 
@@ -128,9 +90,6 @@ public class 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
      */
@@ -162,12 +121,11 @@ public class Queue
         try
         {
             // TODO: Make us wait on the event (optional with a time-out)
-            signal.notifyAll();
+            event.notifyAll();
         }
-        catch(SyncError snozErr)
+        catch(SnoozeError snozErr)
         {
-            // Throw an exception on a fatal exception
-            throw new TristanableException(ErrorType.ENQUEUE_FAILED);
+            // TODO: Add error handling for libsnooze exceptions here
         }
     }
 
@@ -176,9 +134,6 @@ public class Queue
     /** 
      * 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()
@@ -202,26 +157,16 @@ public class Queue
         /* Block till we dequeue a message successfully */
         while(dequeuedMessage is null)
         {
-            scope(exit)
-            {
-                // Unlock the mutex
-                this.mutex.unlock();
-            }
-
-            // Lock the mutex
-            this.mutex.lock();
-
             try
             {
-                this.signal.wait(this.wakeInterval);
+                // TODO: Make us wait on the event (optional with a time-out)
+                event.wait();
             }
-            catch(SyncError e)
+            catch(SnoozeError snozErr)
             {
-                // Throw an exception on a fatal exception
-                throw new TristanableException(ErrorType.DEQUEUE_FAILED);
+                // TODO: Add error handling for libsnooze exceptions here
             }
 
-
             /* Lock the item queue */
             queueLock.lock();