Merge branch 'nextgen' into nextgen_listener
This commit is contained in:
commit
c5c01c32e1
|
|
@ -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}}
|
||||
|
|
|
|||
|
|
@ -3,7 +3,8 @@
|
|||
tristanable
|
||||
===========
|
||||
|
||||
[](https://github.com/deavmi/tristanable/actions/workflows/d.yml)
|
||||
[](https://github.com/deavmi/tristanable/actions/workflows/d.yml)    [](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`.
|
||||
|
||||
|
|
|
|||
3
dub.json
3
dub.json
|
|
@ -4,8 +4,7 @@
|
|||
],
|
||||
"copyright": "Copyright © 2023, Tristan B. Kildaire",
|
||||
"dependencies": {
|
||||
"bformat": "4.1.0",
|
||||
"libsnooze": "0.3.3"
|
||||
"bformat": ">=4.1.1"
|
||||
},
|
||||
"description": "Tristanable network message queuing framework",
|
||||
"homepage": "https://deavmi.assigned.network/projects/tristanable",
|
||||
|
|
|
|||
|
|
@ -24,7 +24,17 @@ public enum ErrorType
|
|||
/**
|
||||
* If no default queue is configured
|
||||
*/
|
||||
NO_DEFAULT_QUEUE
|
||||
NO_DEFAULT_QUEUE,
|
||||
|
||||
/**
|
||||
* The blocking call to `dequeue()`, somehow, failed
|
||||
*/
|
||||
DEQUEUE_FAILED,
|
||||
|
||||
/**
|
||||
* The call to `enqueue()`, somehow, failed
|
||||
*/
|
||||
ENQUEUE_FAILED
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -5,14 +5,13 @@ 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.thread : dur;
|
||||
import core.time : Duration, dur;
|
||||
import tristanable.exceptions;
|
||||
|
||||
version(unittest)
|
||||
{
|
||||
|
|
@ -35,8 +34,15 @@ public class Queue
|
|||
/**
|
||||
* 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
|
||||
|
|
@ -60,6 +66,15 @@ 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
|
||||
|
|
@ -75,14 +90,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;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -90,6 +128,9 @@ 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
|
||||
*/
|
||||
|
|
@ -121,11 +162,12 @@ public class Queue
|
|||
try
|
||||
{
|
||||
// TODO: Make us wait on the event (optional with a time-out)
|
||||
event.notifyAll();
|
||||
signal.notifyAll();
|
||||
}
|
||||
catch(SnoozeError snozErr)
|
||||
catch(SyncError snozErr)
|
||||
{
|
||||
// TODO: Add error handling for libsnooze exceptions here
|
||||
// Throw an exception on a fatal exception
|
||||
throw new TristanableException(ErrorType.ENQUEUE_FAILED);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -134,6 +176,9 @@ 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()
|
||||
|
|
@ -157,16 +202,26 @@ 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
|
||||
{
|
||||
// TODO: Make us wait on the event (optional with a time-out)
|
||||
event.wait();
|
||||
this.signal.wait(this.wakeInterval);
|
||||
}
|
||||
catch(SnoozeError snozErr)
|
||||
catch(SyncError e)
|
||||
{
|
||||
// TODO: Add error handling for libsnooze exceptions here
|
||||
// Throw an exception on a fatal exception
|
||||
throw new TristanableException(ErrorType.DEQUEUE_FAILED);
|
||||
}
|
||||
|
||||
|
||||
/* Lock the item queue */
|
||||
queueLock.lock();
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue