Compare commits

...

4 Commits

Author SHA1 Message Date
Tristan B. Velloza Kildaire 42c6b111c8
Migrate from libsnooze (#8)
* Dub

- Removed `libsnooze` dependency

* Queue

- Removed `libsnooze` imports

* Queue

- Added mutex+condition variable

* Queue

- Removed old `ensure()` call

* Queue

- Switched one thing over to mutex+condvar

* Queue

- Switched to using condition variable
- Added configurable slumber interval
2023-10-01 20:51:08 +02:00
Tristan B. Velloza Kildaire 761ddb2d1e Merge branch 'master' into nextgen 2023-10-01 19:14:57 +02:00
Tristan B. Velloza Kildaire a23017a747 README
- Added badges
2023-10-01 19:14:16 +02:00
Tristan B. Velloza Kildaire 459f4a8709 Pipelines
- Added code coverage
2023-10-01 19:10:57 +02:00
4 changed files with 81 additions and 57 deletions

View File

@ -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}}

View File

@ -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`.

View File

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

View File

@ -3,14 +3,12 @@
*/
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.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)
@ -27,10 +25,15 @@ version(unittest)
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
@ -47,6 +50,15 @@ public class Queue
*/
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 +74,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 +146,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 +186,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();