Add MemoryQueueWorker

huntlabs · Apr 7, 2020 · fe434f3 · fe434f3
1 parent a36cdab
commit fe434f3
Show file tree

Hide file tree

Showing 5 changed files with 58 additions and 23 deletions.
diff --git a/.gitignore b/.gitignore
@@ -30,3 +30,4 @@ docs/
 # Others
 /java
 hunt-framework-test-*
+temp.txt
diff --git a/source/hunt/framework/provider/QueueServiceProvider.d b/source/hunt/framework/provider/QueueServiceProvider.d
@@ -25,8 +25,7 @@ class QueueServiceProvider : ServiceProvider {
         string typeName = config.queue.driver;
 
         if (typeName == QueueWorker.Memory) {
-            // _queueWorker = new MemoryQueueWorker();
-            warningf("TODO: %s", typeName);
+            _queueWorker = new MemoryQueueWorker();
         } else if (typeName == QueueWorker.AMQP) {
             auto amqpConf = config.amqp;
 

diff --git a/source/hunt/framework/queue/AmqpQueueWorker.d b/source/hunt/framework/queue/AmqpQueueWorker.d
@@ -17,8 +17,8 @@ class AmqpQueueWorker : QueueWorker {
     private AmqpConnection _listenerConn;
 
     this(AmqpPool pool) {
-        super();
         _pool = pool;
+        super();
     }
 
     override void onListen() {
@@ -38,10 +38,12 @@ class AmqpQueueWorker : QueueWorker {
 
                     recv.handler(new class Handler!AmqpMessage {
                         void handle(AmqpMessage msg) {
-                            tracef("channel: %s Message: %s", channel, msg.bodyAsString());
                             // ubyte[] content = cast(ubyte[])msg.bodyAsBinary();
                             ubyte[] content = cast(ubyte[])msg.bodyAsString();
-                            tracef("%(%02X %)", content);
+                            version(HUNT_FM_DEBUG) {
+                                tracef("channel: %s Message: %s", channel, msg.bodyAsString());
+                                tracef("%(%02X %)", content);
+                            }
                             if(listener !is null) {
                                 listener(content);
                             }

diff --git a/source/hunt/framework/queue/MemoryQueueWorker.d b/source/hunt/framework/queue/MemoryQueueWorker.d
@@ -12,19 +12,52 @@ import std.parallelism;
 /**
  * 
  */
-// class MemoryQueueWorker : QueueWorker {
-//     private SimpleQueue!Job _jobs;
-
-//     this() {
-//         _jobs = new SimpleQueue!Job();
-//         super();
-//     }
-
-//     override protected Job retrieveNextJob() {
-//         return _jobs.dequeue();
-//     }
-
-//     override void push(string channel, ubyte[] message) {
-//         _jobs.enqueue(job);
-//     }
-// }
+class MemoryQueueWorker : QueueWorker {
+    private SimpleQueue!(ubyte[])[string] _queueMap;
+
+    this() {
+        super();
+    }
+
+    override void onListen() {
+        if(listeners is null) {
+            return;
+        }
+
+        foreach(string channel, QueueMessageListener listener; listeners) {
+            auto itemPtr = channel in _queueMap;
+            if(itemPtr is null) {
+                // version(HUNT_DEBUG) warningf("Can't find channel: %s", channel);
+                _queueMap[channel] = new SimpleQueue!(ubyte[])();
+                itemPtr = channel in _queueMap;
+            }
+
+            ubyte[] content = itemPtr.dequeue();
+            version(HUNT_FM_DEBUG) {
+                tracef("channel: %s Message: %s", channel, msg.bodyAsString());
+                tracef("%(%02X %)", content);
+            }
+
+            if(listener !is null) {
+                listener(content);
+            }
+        }
+    }
+
+    override void push(string channel, ubyte[] message) {
+        synchronized(this) {
+            auto itemPtr = channel in _queueMap;
+            if(itemPtr is null) {
+                _queueMap[channel] = new SimpleQueue!(ubyte[])();
+                itemPtr = channel in _queueMap;
+            }
+
+            itemPtr.enqueue(message);
+        }
+    }
+
+    override protected void onStop() {
+        _queueMap.clear();
+    }
+}
+
diff --git a/source/hunt/framework/queue/package.d b/source/hunt/framework/queue/package.d
@@ -1,5 +1,5 @@
 module hunt.framework.queue;
 
-public import hunt.framework.queue.Job;
-public import hunt.framework.queue.QueueWorker;
 public import hunt.framework.queue.AmqpQueueWorker;
+public import hunt.framework.queue.QueueWorker;
+public import hunt.framework.queue.MemoryQueueWorker;