Queue and multi fiber (#5)

* Changed to fiber safe approach that supports queue of events

* Forgot some things

Author: elliotize

README.md

@@ -74,6 +74,13 @@ Spectrum.dispatch_async_event(event_one, :my_queue)
 event_two = FooBarTwoEvent.new(id: 2)
 Spectrum.dispatch_async_event(event_two, :my_queue)
 Spectrum::EventQueue.stop(:my_queue)
+
+## Advanced Queue Settings
+Spectrum::EventQueue.start(
+  name: :my_queue,
+  number_of_workers: 10
+  queue_size: 10
+)
 ```
 
 ## Contributing

shard.yml

@@ -1,5 +1,5 @@
 name: spectrum
-version: 0.1.0
+version: 0.2.0
 
 authors:
   - Tom Elliot <telliot16@gmail.com>

spec/event_spec.cr

@@ -1,35 +1,95 @@
-require "./spec_helper"
-include Spectrum
-describe Spectrum do
-  it "should have only events that are listening to an event recieve it" do
-    event_one = FooBarOneEvent.new(id: 1)
-    event_two = FooBarTwoEvent.new(id: 2)
-    event_three = FooBarOneEvent.new(id: 3)
-
-    Spectrum.dispatch_sync_event(event_one)
-    Spectrum.dispatch_sync_event(event_two)
-    Spectrum.dispatch_sync_event(event_three)
-
-    event_one.handlers.size.should eq(2)
-    event_two.handlers.size.should eq(2)
-    event_three.handlers.size.should eq(2)
-  end
-
-  it "async events should be triggered" do
-    event_queue = EventQueue.start(:test)
-    event_one = FooBarOneEvent.new(id: 1)
-    event_two = FooBarTwoEvent.new(id: 2)
-    event_three = FooBarOneEvent.new(id: 3)
-
-    Spectrum.dispatch_async_event(event_one, :test)
-    Spectrum.dispatch_async_event(event_two, :test)
-    Spectrum.dispatch_async_event(event_three, :test)
-    while event_queue.busy?
-      # This is just a loop to ensure there is not a race condition
-    end
-    event_one.handlers.size.should eq(2)
-    event_two.handlers.size.should eq(2)
-    event_three.handlers.size.should eq(2)
-    EventQueue.stop(:test)
-  end
-end
+require "./spec_helper"
+include Spectrum
+describe Spectrum do
+  it "should have only events that are listening to an event recieve it" do
+    event_one = FooBarOneEvent.new(id: 1)
+    event_two = FooBarTwoEvent.new(id: 2)
+    event_three = FooBarOneEvent.new(id: 3)
+
+    Spectrum.dispatch_sync_event(event_one)
+    Spectrum.dispatch_sync_event(event_two)
+    Spectrum.dispatch_sync_event(event_three)
+
+    event_one.handlers.size.should eq(2)
+    event_two.handlers.size.should eq(2)
+    event_three.handlers.size.should eq(2)
+  end
+
+  it "async events should be triggered" do
+    event_queue = EventQueue.start(:test)
+    event_one = FooBarOneEvent.new(id: 1)
+    event_two = FooBarTwoEvent.new(id: 2)
+    event_three = FooBarOneEvent.new(id: 3)
+
+    Spectrum.dispatch_async_event(event_one, :test)
+    Spectrum.dispatch_async_event(event_two, :test)
+    Spectrum.dispatch_async_event(event_three, :test)
+
+    Fiber.yield
+
+    event_one.handlers.size.should eq(2)
+    event_two.handlers.size.should eq(2)
+    event_three.handlers.size.should eq(2)
+    EventQueue.stop(:test)
+  end
+
+  it "async events should be none blocking for the caller" do
+    event_queue1 = EventQueue.start(:test1)
+    event_queue2 = EventQueue.start(:test2)
+    event1 = LockEvent.new
+    event2 = LockEvent.new
+    Spectrum.dispatch_async_event(event1, :test1)
+    Spectrum.dispatch_async_event(event1, :test1)
+    Spectrum.dispatch_async_event(event2, :test2)
+    Spectrum.dispatch_async_event(event2, :test2)
+    Fiber.yield
+    event1.unlock!
+    event2.unlock!
+    Fiber.yield
+    event1.unlock!
+    event2.unlock!
+    EventQueue.stop(:test1)
+    EventQueue.stop(:test2)
+  end
+
+  it "async events should queue" do
+    event_queue = EventQueue.start(name: :test, number_of_workers: 1, queue_size: 2)
+    event1 = LockEvent.new
+    event2 = LockEvent.new
+    event3 = LockEvent.new
+    Spectrum.dispatch_async_event(event1, :test)
+    Spectrum.dispatch_async_event(event2, :test)
+    Spectrum.dispatch_async_event(event3, :test)
+    Fiber.yield
+    event1.unlock!
+    event2.unlock!
+    event3.unlock!
+    Fiber.yield
+    event1.unlock!
+    event2.unlock!
+    event3.unlock!
+    EventQueue.stop(:test)
+  end
+
+  it "async events should queue hit error if queue is full and timeout is hit" do
+    event_queue = EventQueue.start(name: :test, number_of_workers: 1, queue_size: 1, queue_timeout: 0)
+    event1 = LockEvent.new
+    event2 = LockEvent.new
+    event3 = LockEvent.new
+    Spectrum.dispatch_async_event(event1, :test)
+    Fiber.yield
+    Spectrum.dispatch_async_event(event2, :test)
+    Fiber.yield
+    error = nil
+    begin
+      Spectrum.dispatch_async_event(event3, :test)
+    rescue e : Spectrum::EventQueue::FullException
+      error = e
+    end
+    event1.unlock!
+    event2.unlock!
+    event3.unlock!
+    error.class.should eq(Spectrum::EventQueue::FullException)
+    EventQueue.stop(:test)
+  end
+end

spec/spec_helper.cr

@@ -1,57 +1,85 @@
-require "spec"
-require "../src/spectrum"
-
-class FooBarOneEvent
-  include Spectrum::Event
-  getter :handlers
-  def initialize(@id : Int32)
-    @handlers = [] of Spectrum::EventHandler
-  end
-
-  def store_handler(handler : Spectrum::EventHandler)
-    @handlers << handler
-  end
-end
-
-class FooBarTwoEvent
-  include Spectrum::Event
-  getter :handlers
-  def initialize(@id : Int32)
-    @handlers = [] of Spectrum::EventHandler
-  end
-
-  def store_handler(handler : Spectrum::EventHandler)
-    @handlers << handler
-  end
-end
-
-class FooBarOneEventHandler
-  include Spectrum::EventHandler
-  Syringe.injectable
-
-  def on_event(event : FooBarOneEvent)
-    event.store_handler(self)
-  end
-end
-
-class FooBarTwoEventHandler
-  include Spectrum::EventHandler
-  Syringe.injectable
-
-  def on_event(event : FooBarTwoEvent)
-    event.store_handler(self)
-  end
-end
-
-class FooBarBothEventHandler
-  include Spectrum::EventHandler
-  Syringe.injectable
-
-  def on_event(event : FooBarOneEvent)
-    event.store_handler(self)
-  end
-
-  def on_event(event : FooBarTwoEvent)
-    event.store_handler(self)
-  end
-end
+require "spec"
+require "../src/spectrum"
+
+class FooBarOneEvent
+  include Spectrum::Event
+  getter :handlers
+  def initialize(@id : Int32)
+    @handlers = [] of Spectrum::EventHandler
+  end
+
+  def store_handler(handler : Spectrum::EventHandler)
+    @handlers << handler
+  end
+end
+
+class FooBarTwoEvent
+  include Spectrum::Event
+  getter :handlers
+  def initialize(@id : Int32)
+    @handlers = [] of Spectrum::EventHandler
+  end
+
+  def store_handler(handler : Spectrum::EventHandler)
+    @handlers << handler
+  end
+end
+
+class FooBarOneEventHandler
+  include Spectrum::EventHandler
+  Syringe.injectable
+
+  def on_event(event : FooBarOneEvent)
+    event.store_handler(self)
+  end
+end
+
+class FooBarTwoEventHandler
+  include Spectrum::EventHandler
+  Syringe.injectable
+
+  def on_event(event : FooBarTwoEvent)
+    event.store_handler(self)
+  end
+end
+
+class FooBarBothEventHandler
+  include Spectrum::EventHandler
+  Syringe.injectable
+
+  def on_event(event : FooBarOneEvent)
+    event.store_handler(self)
+  end
+
+  def on_event(event : FooBarTwoEvent)
+    event.store_handler(self)
+  end
+end
+
+class LockEvent
+  include Spectrum::Event
+  getter :lock
+  def initialize
+    @lock = false
+  end
+
+  def lock!
+    @lock = true
+  end
+
+  def unlock!
+    @lock = false
+  end
+end
+
+class LockEventHandler
+  include Spectrum::EventHandler
+  Syringe.injectable
+
+  def on_event(event : LockEvent)
+    event.lock!
+    while event.lock
+      Fiber.yield
+    end
+  end
+end

src/spectrum/event_dispatcher.cr

@@ -1,13 +1,13 @@
-module Spectrum
-  class EventDispatcher
-    def dispatch_sync_event(event : Event)
-      listeners = EventHandlers.new
-      listeners.on_event(event)
-    end
-
-    def dispatch_async_event(event : Event, queue_name : Symbol)
-      event_queue = EventQueue.get(queue_name)
-      event_queue.send(event)
-    end
-  end
+module Spectrum
+  class EventDispatcher
+    def dispatch_sync_event(event : Event)
+      listeners = EventHandlers.new
+      listeners.on_event(event)
+    end
+
+    def dispatch_async_event(event : Event, queue_name : Symbol)
+      event_queue = EventQueue.get(queue_name)
+      event_queue.send(event)
+    end
+  end
 end

src/spectrum/event_handler.cr

@@ -1,7 +1,7 @@
-module Spectrum
-  module EventHandler
-    def on_event(event : Event)
-      # This is just a hook that is overloaded
-    end
-  end
+module Spectrum
+  module EventHandler
+    def on_event(event : Event)
+      # This is just a hook that is overloaded
+    end
+  end
 end

src/spectrum/event_handlers.cr

@@ -1,16 +1,16 @@
-require "syringe"
-module Spectrum
-  class EventHandlers
-    include Syringe
-    Syringe.injectable
-
-    def initialize(@handlers : Array(EventHandler))
-    end
-
-    def on_event(event : Event)
-      @handlers.each do |handler|
-        handler.on_event(event)
-      end
-    end
-  end
+require "syringe"
+module Spectrum
+  class EventHandlers
+    include Syringe
+    Syringe.injectable
+
+    def initialize(@handlers : Array(EventHandler))
+    end
+
+    def on_event(event : Event)
+      @handlers.each do |handler|
+        handler.on_event(event)
+      end
+    end
+  end
 end