2.0.0

README.md

@@ -45,5 +45,13 @@ bus.push(topic "Dogs", "dog-related message")
 bus.push(toIntSet([topic "Cats", topic "Dogs"]), "multitopical message")
 
 bus.doDelivery()
-jointhreads bus.stop()
+bus.stop()
 ```
+
+
+## Release notes for 2.0.0 (2021-08-09)
+- Alternative *newSuber* procs removed, use the one remaining
+- Push, Pull and Deliver callbacks must be registered via corresponding *setXXXCallback* procs
+- new *doSynced* proc for running code in sync with message sending
+- *stop* does not return a thread, joins the thread automatically instead
+- other bug fixes and minor improvements

suber.nimble

@@ -1,6 +1,6 @@
 # Package
 
-version       = "1.0.0"
+version       = "2.0.0"
 author        = "Olli"
 description   = "Pub/Sub engine"
 license       = "MIT"

tests/test_delivery.nim

@@ -20,10 +20,16 @@ proc onBDeliver(messages: openArray[ptr SuberMessage[int]]) =
   {.gcsafe.}:
     for m in messages: bDeliveredmessages.incl(m.data)
 
-let a = newSuber[int, TopicCount](onADeliver)
-let b = newSuber[int, TopicCount](onBDeliver, 10000, 20000, 10000, 50)
+let a = newSuber[int, TopicCount]()
+let b = newSuber[int, TopicCount](10000, 20000, 10000, 50)
 
 var lock: Lock
+lock.initLock
+
+a.setDeliverCallback(onADeliver)
+b.setDeliverCallback(onBDeliver)
+a.subscribe(1.Subscriber, 1.Topic, true)
+b.subscribe(1.Subscriber, 1.Topic, true)
 
 proc run(t: int) =
   {.gcsafe.}:
@@ -36,16 +42,15 @@ proc run(t: int) =
       b.push(1.Topic, data, 1)
       withLock(lock): pushedmessages.incl(data)
 
+
 var threads: array[ThreadCount, Thread[int]]
-a.subscribe(1.Subscriber, 1.Topic, true)
-b.subscribe(1.Subscriber, 1.Topic, true)
-lock.initLock
 echo "Multi-threaded delivery testing with ", ThreadCount * DeliveryCount, " messages"
 for i in 0 ..< ThreadCount: createThread(threads[i], run, i)
 joinThreads(threads)
 a.doDelivery()
 b.doDelivery()
-joinThreads([a.stop(), b.stop()])
+a.stop()
+b.stop()
 doAssert(pushedmessages.len == ThreadCount * DeliveryCount)
 doAssert(pushedmessages.len == aDeliveredmessages.len)
 doAssert(aDeliveredmessages.len == bDeliveredmessages.len)
@@ -55,7 +60,7 @@ echo "b Max channel queue length: ", b.getChannelQueueLengths[2]
 sleep(1000)
 
 #--------------------------
-# gc:orc may SIGSEGV here, it is not a Suber bug!
+# gc:orc may SIGSEGV here, it is not a Suber bug
 pushedmessages.clear()
 aDeliveredmessages.clear()
 bDeliveredmessages.clear()

tests/test_generic.nim

@@ -36,12 +36,17 @@ proc onADeliver(messages: openArray[ptr Message]) = aDeliveredmessages += messag
 
 proc onBDeliver(messages: openArray[ptr Message]) {.gcsafe, raises:[].} = bDeliveredmessages += messages.len
 
-let a: Suber[MessageData, MaxTopics] = newSuber[MessageData, MaxTopics](onAPush, onADeliver)
+let a: Suber[MessageData, MaxTopics] = newSuber[MessageData, MaxTopics]()
 
-let b: Suber[MessageData, MaxTopics] = newSuber[MessageData, MaxTopics](onBPush, onBDeliver, 1000, 20, 5, 10)
+let b: Suber[MessageData, MaxTopics] = newSuber[MessageData, MaxTopics](1000, 20, 5, 10)
 
 var rounds = 0
 
+a.setPushCallback(onAPush)
+b.setPushCallback(onBpush)
+a.setDeliverCallback(onADeliver)
+b.setDeliverCallback(onBDeliver)
+
 proc addTopic() =
   if a.getTopiccount() == MaxTopics: return
   addtopics.inc
@@ -68,7 +73,6 @@ proc removeTopic() =
     if a.getSubscriptions(subscribers[s]).len == 0: nomores.add(subscribers[s])
   for n in nomores: subscribers.del(subscribers.find(n))
 
-from os import sleep
 proc addSubscriber() =
   addsubscribers.inc
   if topics.len == 0: addTopic()
@@ -121,8 +125,9 @@ proc run() =
   b.doDelivery()
 
 run()
-joinThread a.stop()
-joinThread b.stop()
+a.removeTopic(255)
+a.stop()
+b.stop()
 doAssert(publishedmessages > 1000)
 doAssert(publishedmessages == aPushedmessages)
 doAssert(aPushedmessages == bPushedmessages)

tests/test_speed.nim

@@ -10,7 +10,8 @@ var messagecount: int
 
 proc onDeliver(messages: openArray[ptr SuberMessage[int]]) = {.gcsafe.}: discard messagecount.atomicInc(messages.len)
 
-let bus = newSuber[int, 1](onDeliver, 1000000, 100000, 100)
+let bus = newSuber[int, 1](1000000, 100000, 100)
+bus.setDeliverCallback(onDeliver)
 bus.subscribe(1.Subscriber, 1.Topic, true)
 
 var stop: bool
@@ -26,5 +27,5 @@ for i in 0 ..< ThreadCount: createThread(threads[i], run)
 sleep TestDuration * 1000
 stop = true
 joinThreads(threads)
-joinThread bus.stop()
+bus.stop()
 echo messagecount div TestDuration, " msg/s"