Compare commits
4 Commits
42c6b111c8
...
aced8fdc76
Author | SHA1 | Date |
---|---|---|
Tristan B. Velloza Kildaire | aced8fdc76 | |
Tristan B. Velloza Kildaire | dafcdf714b | |
Tristan B. Velloza Kildaire | 5cafbb8130 | |
Tristan B. Velloza Kildaire | be91f5ffc5 |
3
dub.json
3
dub.json
|
@ -4,7 +4,8 @@
|
||||||
],
|
],
|
||||||
"copyright": "Copyright © 2023, Tristan B. Kildaire",
|
"copyright": "Copyright © 2023, Tristan B. Kildaire",
|
||||||
"dependencies": {
|
"dependencies": {
|
||||||
"bformat": ">=4.1.1"
|
"bformat": ">=4.1.1",
|
||||||
|
"niknaks": ">=0.3.0"
|
||||||
},
|
},
|
||||||
"description": "Tristanable network message queuing framework",
|
"description": "Tristanable network message queuing framework",
|
||||||
"homepage": "https://deavmi.assigned.network/projects/tristanable",
|
"homepage": "https://deavmi.assigned.network/projects/tristanable",
|
||||||
|
|
|
@ -4,6 +4,7 @@
|
||||||
module tristanable.encoding;
|
module tristanable.encoding;
|
||||||
|
|
||||||
import std.conv : to;
|
import std.conv : to;
|
||||||
|
import niknaks.bits : bytesToIntegral, Order, order, toBytes;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Represents a tagged message that has been decoded
|
* Represents a tagged message that has been decoded
|
||||||
|
@ -60,31 +61,9 @@ public final class TaggedMessage
|
||||||
/* The decoded tag */
|
/* The decoded tag */
|
||||||
ulong decodedTag;
|
ulong decodedTag;
|
||||||
|
|
||||||
/* If on little endian then dump direct */
|
/* Take ulong-many bytes and only flip them to LE if not on LE host */
|
||||||
version(LittleEndian)
|
decodedTag = order(bytesToIntegral!(ushort)(cast(ubyte[])encodedMessage), Order.LE);
|
||||||
{
|
|
||||||
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 */
|
/* Set the tag */
|
||||||
decodedMessage.setTag(decodedTag);
|
decodedMessage.setTag(decodedTag);
|
||||||
|
@ -106,41 +85,9 @@ public final class TaggedMessage
|
||||||
/* The encoded bytes */
|
/* The encoded bytes */
|
||||||
byte[] encodedMessage;
|
byte[] encodedMessage;
|
||||||
|
|
||||||
/* If on little endian, then dump 64 bit as is - little endian */
|
/* If on little endian then no re-order, if host is BE flip (the tag) */
|
||||||
version(LittleEndian)
|
encodedMessage ~= toBytes(order(tag, Order.LE));
|
||||||
{
|
|
||||||
/* 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 */
|
/* Tack on the data */
|
||||||
encodedMessage ~= data;
|
encodedMessage ~= data;
|
||||||
|
|
|
@ -4,7 +4,7 @@
|
||||||
module tristanable.manager.manager;
|
module tristanable.manager.manager;
|
||||||
|
|
||||||
import std.socket;
|
import std.socket;
|
||||||
import tristanable.queue : Queue;
|
import tristanable.queue.queue : Queue;
|
||||||
import core.sync.mutex : Mutex;
|
import core.sync.mutex : Mutex;
|
||||||
import tristanable.manager.watcher : Watcher;
|
import tristanable.manager.watcher : Watcher;
|
||||||
import tristanable.encoding : TaggedMessage;
|
import tristanable.encoding : TaggedMessage;
|
||||||
|
|
|
@ -11,7 +11,7 @@ import std.socket;
|
||||||
import bformat;
|
import bformat;
|
||||||
import tristanable.encoding;
|
import tristanable.encoding;
|
||||||
import tristanable.exceptions;
|
import tristanable.exceptions;
|
||||||
import tristanable.queue;
|
import tristanable.queue.queue;
|
||||||
import bformat.client;
|
import bformat.client;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -204,6 +204,28 @@ unittest
|
||||||
writeln("server send status: ", bClient.sendMessage(tEncoded));
|
writeln("server send status: ", bClient.sendMessage(tEncoded));
|
||||||
|
|
||||||
writeln("server send [done]");
|
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]");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -223,6 +245,10 @@ unittest
|
||||||
manager.registerQueue(sixtyNine);
|
manager.registerQueue(sixtyNine);
|
||||||
manager.registerQueue(fortyTwo);
|
manager.registerQueue(fortyTwo);
|
||||||
|
|
||||||
|
// Register a default queue (tag ignored)
|
||||||
|
Queue defaultQueue = new Queue(2332);
|
||||||
|
manager.setDefaultQueue(defaultQueue);
|
||||||
|
|
||||||
|
|
||||||
/* Connect our socket to the server */
|
/* Connect our socket to the server */
|
||||||
client.connect(server.localAddress);
|
client.connect(server.localAddress);
|
||||||
|
@ -252,6 +278,19 @@ unittest
|
||||||
assert(dequeuedMessage.getPayload() == cast(byte[])"Cucumber 😳️");
|
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 */
|
/* Stop the manager */
|
||||||
manager.stop();
|
manager.stop();
|
||||||
|
|
|
@ -12,7 +12,7 @@ public import tristanable.manager;
|
||||||
/**
|
/**
|
||||||
* A queue of queue items all of the same tag
|
* A queue of queue items all of the same tag
|
||||||
*/
|
*/
|
||||||
public import tristanable.queue : Queue;
|
public import tristanable.queue.queue : Queue;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Error handling type definitions
|
* Error handling type definitions
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
/**
|
/**
|
||||||
* A queue of queue items all of the same tag
|
* A queue of queue items all of the same tag
|
||||||
*/
|
*/
|
||||||
module tristanable.queue;
|
module tristanable.queue.queue;
|
||||||
|
|
||||||
import core.sync.mutex : Mutex;
|
import core.sync.mutex : Mutex;
|
||||||
import core.sync.condition : Condition;
|
import core.sync.condition : Condition;
|
||||||
|
@ -25,6 +25,13 @@ version(unittest)
|
||||||
public class Queue
|
public class Queue
|
||||||
{
|
{
|
||||||
/**
|
/**
|
||||||
|
* This queue's unique ID
|
||||||
|
*/
|
||||||
|
private ulong queueID;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* The libsnooze event used to sleep/wake
|
||||||
|
* on queue events
|
||||||
* Mutex for the condition variable
|
* Mutex for the condition variable
|
||||||
*/
|
*/
|
||||||
private Mutex mutex;
|
private Mutex mutex;
|
||||||
|
@ -44,13 +51,8 @@ public class Queue
|
||||||
* The lock for the message queue
|
* The lock for the message queue
|
||||||
*/
|
*/
|
||||||
private Mutex queueLock;
|
private Mutex queueLock;
|
||||||
|
|
||||||
/**
|
|
||||||
* This queue's unique ID
|
|
||||||
*/
|
|
||||||
private ulong queueID;
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* If a message is enqueued prior
|
* If a message is enqueued prior
|
||||||
* to us sleeping then we won't
|
* to us sleeping then we won't
|
||||||
* wake up and return for it.
|
* wake up and return for it.
|
Loading…
Reference in New Issue