Compare commits
4 Commits
v3.2.4-bet
...
master
Author | SHA1 | Date |
---|---|---|
Tristan B. Velloza Kildaire | 4bee62e4e2 | |
Tristan B. Velloza Kildaire | b24c3368b3 | |
Tristan B. Velloza Kildaire | a23017a747 | |
Tristan B. Velloza Kildaire | 459f4a8709 |
|
@ -6,9 +6,9 @@ name: D
|
|||
|
||||
on:
|
||||
push:
|
||||
branches: [ "master" ]
|
||||
branches: [ "**" ]
|
||||
pull_request:
|
||||
branches: [ "master", "nextgen" ]
|
||||
branches: [ "**" ]
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
@ -22,6 +22,11 @@ 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
|
||||
|
@ -29,4 +34,8 @@ 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
|
||||
dub test --compiler=$DC --coverage
|
||||
|
||||
- name: Coverage upload
|
||||
run: |
|
||||
dub run doveralls -- -t ${{secrets.COVERALLS_REPO_TOKEN}}
|
||||
|
|
166
README.md
166
README.md
|
@ -3,7 +3,8 @@
|
|||
tristanable
|
||||
===========
|
||||
|
||||
[![D](https://github.com/deavmi/tristanable/actions/workflows/d.yml/badge.svg)](https://github.com/deavmi/tristanable/actions/workflows/d.yml)
|
||||
[![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 respectively tagged queues indicated by their _"type"_ or `id`.
|
||||
|
||||
|
@ -17,44 +18,155 @@ Tristanable provides a way for you to receive the "IM notification first" but bl
|
|||
|
||||
### 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 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.
|
||||
Below is a fully-fledged example of the types of places (networking) where tristanable can be of help. The code is all explained in the comments:
|
||||
|
||||
```d
|
||||
import tristanable.manager;
|
||||
import tristanable.queue;
|
||||
import tristanable.queueitem;
|
||||
import std.socket;
|
||||
import std.stdio;
|
||||
import core.thread;
|
||||
|
||||
/* Create a manager to manage the socket for us */
|
||||
Manager manager = new Manager(socket);
|
||||
Address serverAddress = parseAddress("::1", 0);
|
||||
Socket server = new Socket(AddressFamily.INET6, SocketType.STREAM, ProtocolType.TCP);
|
||||
server.bind(serverAddress);
|
||||
server.listen(0);
|
||||
|
||||
/* Create a Queue for all "weather messages" */
|
||||
Queue weatherQueue = new Queue(1);
|
||||
class ServerThread : Thread
|
||||
{
|
||||
this()
|
||||
{
|
||||
super(&worker);
|
||||
}
|
||||
|
||||
/* Create a Queue for all "IM notifications" */
|
||||
Queue instantNotification = new Queue(2);
|
||||
private void worker()
|
||||
{
|
||||
Socket clientSocket = server.accept();
|
||||
BClient bClient = new BClient(clientSocket);
|
||||
|
||||
/* Tell the manager to look out for tagged messages `1` and `2` */
|
||||
manager.addQueue(weatherQueue);
|
||||
manager.addQueue(instantNotification);
|
||||
Thread.sleep(dur!("seconds")(7));
|
||||
writeln("Server start");
|
||||
|
||||
/* Now we can block on this queue and return with its head */
|
||||
QueueItem message = weatherQueue.dequeue();
|
||||
```
|
||||
/**
|
||||
* 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));
|
||||
|
||||
Surely, there must be some sort of encoding mechanism too? The messages after all need to be encoded. **No problem!**, we have that sorted:
|
||||
writeln("server send [done]");
|
||||
|
||||
```d
|
||||
import tristanable.encoding;
|
||||
/**
|
||||
* 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));
|
||||
|
||||
/* Let's send it with tag 1 and data "Hello" */
|
||||
ulong tag = 1;
|
||||
byte[] data = cast(byte[])"Hello";
|
||||
writeln("server send [done]");
|
||||
|
||||
/* When sending a message */
|
||||
DataMessage tristanEncoded = new DataMessage(tag, data);
|
||||
/**
|
||||
* 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));
|
||||
|
||||
/* Then send it */
|
||||
socket.send(encodeForSend(tristanEncoded));
|
||||
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]");
|
||||
}
|
||||
}
|
||||
|
||||
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);
|
||||
|
||||
// Register a default queue (tag ignored)
|
||||
Queue defaultQueue = new Queue(2332);
|
||||
manager.setDefaultQueue(defaultQueue);
|
||||
|
||||
|
||||
/* 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 😳️");
|
||||
|
||||
|
||||
/* 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();
|
||||
```
|
||||
|
||||
And let tristanable handle it! We even handle the message lengths and everything using another great project [bformat](https://deavmi.assigned.network/projects/bformat).
|
||||
|
|
2
dub.json
2
dub.json
|
@ -5,7 +5,7 @@
|
|||
"copyright": "Copyright © 2023, Tristan B. Kildaire",
|
||||
"dependencies": {
|
||||
"bformat": ">=4.1.1",
|
||||
"libsnooze": ">=1.3.0-beta"
|
||||
"niknaks": ">=0.3.0"
|
||||
},
|
||||
"description": "Tristanable network message queuing framework",
|
||||
"homepage": "https://deavmi.assigned.network/projects/tristanable",
|
||||
|
|
|
@ -4,6 +4,7 @@
|
|||
module tristanable.encoding;
|
||||
|
||||
import std.conv : to;
|
||||
import niknaks.bits : bytesToIntegral, Order, order, toBytes;
|
||||
|
||||
/**
|
||||
* Represents a tagged message that has been decoded
|
||||
|
@ -60,31 +61,9 @@ public final class 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 😳️");
|
||||
}
|
||||
/* 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);
|
||||
|
@ -106,41 +85,9 @@ public final class TaggedMessage
|
|||
/* 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;
|
||||
/* If on little endian then no re-order, if host is BE flip (the tag) */
|
||||
encodedMessage ~= toBytes(order(tag, Order.LE));
|
||||
|
||||
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;
|
||||
|
|
|
@ -4,7 +4,7 @@
|
|||
module tristanable.manager.manager;
|
||||
|
||||
import std.socket;
|
||||
import tristanable.queue : Queue;
|
||||
import tristanable.queue.queue : Queue;
|
||||
import core.sync.mutex : Mutex;
|
||||
import tristanable.manager.watcher : Watcher;
|
||||
import tristanable.encoding : TaggedMessage;
|
||||
|
|
|
@ -11,7 +11,7 @@ import std.socket;
|
|||
import bformat;
|
||||
import tristanable.encoding;
|
||||
import tristanable.exceptions;
|
||||
import tristanable.queue;
|
||||
import tristanable.queue.queue;
|
||||
import bformat.client;
|
||||
|
||||
/**
|
||||
|
@ -204,6 +204,28 @@ 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]");
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -223,6 +245,10 @@ 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);
|
||||
|
@ -252,6 +278,19 @@ 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();
|
||||
|
|
|
@ -12,7 +12,7 @@ public import tristanable.manager;
|
|||
/**
|
||||
* 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
|
||||
|
|
|
@ -1,16 +1,14 @@
|
|||
/**
|
||||
* 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;
|
||||
module tristanable.queue.queue;
|
||||
|
||||
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.thread : dur;
|
||||
import core.time : Duration, dur;
|
||||
import tristanable.exceptions;
|
||||
|
||||
version(unittest)
|
||||
|
@ -26,11 +24,23 @@ version(unittest)
|
|||
*/
|
||||
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
|
||||
*/
|
||||
private Event event;
|
||||
private Mutex mutex;
|
||||
|
||||
/**
|
||||
* The condition variable used to sleep/wake
|
||||
* on queue of events
|
||||
*/
|
||||
private Condition signal;
|
||||
|
||||
/**
|
||||
* The queue of messages
|
||||
|
@ -41,12 +51,16 @@ public class Queue
|
|||
* The lock for the message queue
|
||||
*/
|
||||
private Mutex queueLock;
|
||||
|
||||
/**
|
||||
* This queue's unique ID
|
||||
*/
|
||||
private ulong queueID;
|
||||
|
||||
/**
|
||||
* 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
|
||||
|
@ -62,14 +76,37 @@ public class Queue
|
|||
/* Initialize the queue lock */
|
||||
this.queueLock = new Mutex();
|
||||
|
||||
/* Initialize the event */
|
||||
this.event = new Event();
|
||||
/* Initialize the condition variable */
|
||||
this.mutex = new Mutex();
|
||||
this.signal = new Condition(this.mutex);
|
||||
|
||||
/* 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));
|
||||
/* 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;
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -111,9 +148,9 @@ public class Queue
|
|||
try
|
||||
{
|
||||
// TODO: Make us wait on the event (optional with a time-out)
|
||||
event.notifyAll();
|
||||
signal.notifyAll();
|
||||
}
|
||||
catch(FatalException snozErr)
|
||||
catch(SyncError snozErr)
|
||||
{
|
||||
// Throw an exception on a fatal exception
|
||||
throw new TristanableException(ErrorType.ENQUEUE_FAILED);
|
||||
|
@ -151,45 +188,25 @@ public class Queue
|
|||
/* 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)
|
||||
scope(exit)
|
||||
{
|
||||
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;
|
||||
// Unlock the mutex
|
||||
this.mutex.unlock();
|
||||
}
|
||||
|
||||
|
||||
// Lock the mutex
|
||||
this.mutex.lock();
|
||||
|
||||
try
|
||||
{
|
||||
this.signal.wait(this.wakeInterval);
|
||||
}
|
||||
catch(SyncError e)
|
||||
{
|
||||
// Throw an exception on a fatal exception
|
||||
throw new TristanableException(ErrorType.DEQUEUE_FAILED);
|
||||
}
|
||||
|
||||
|
||||
/* Lock the item queue */
|
||||
queueLock.lock();
|
Loading…
Reference in New Issue