A Pub/Sub engine.

Receives messages from multiple sources and delivers them as a serialized stream. Publisher threads do not block each other or delivery thread, and delivery does not block publishing. Messages can belong to multiple topics. Subscribers can subscribe to multiple topics. Topics, subscribers and subscriptions can be modified anytime. Delivery may be triggered on message push, size of undelivered messages, amount of undelivered messages and on call. Keeps a message cache so that subscribers can sync their state at will.

Example

# nim c -r --gc:arc --threads:on --d:release example.nim
import suber, os, random, times

let topics = ["Art", "Science", "Nim", "Fishing"]
let subscribers = ["Amy", "Bob", "Chas", "Dave"]
let messagedatas = ["Good News", "Bad News", "Breaking News", "Old News"]
let publishers = ["Helsinki", "Tallinn", "Oslo", "Edinburgh"]
let bus = newSuber[string, 4]() # string datas, max 4 topics
var stop: bool

proc generateMessages(publisher: int) =
  {.gcsafe.}:
    while true:
      if stop: break else: (sleep(100+rand(500)) ; if stop: break)
      write(stdout, $publishers[publisher] & ", "); flushFile(stdout)
      var messagetopics = initIntSet()
      for topicnumber in 1 .. 1 + rand(2): messagetopics.incl(rand(3))
      bus.push(messagetopics, messagedatas[rand(3)] & " from " & publishers[publisher])

proc deliver() =
  {.gcsafe.}:
    for i in 1 .. 10: (sleep(1000); echo ""; bus.doDelivery())
    stop = true

proc writeMessages(subscriber: int, messages: openArray[ptr SuberMessage[string]]) =
  var subscriberids: IntSet
  try:
    for message in messages:
      if subscriber > -1:
        stdout.write("@" & $message.timestamp & " to " & subscribers[subscriber] & ", ")
      else:
        stdout.write("@" & $message.timestamp & " to ")
        bus.getSubscribers(message, subscriberids)
        for subscriberid in subscriberids: stdout.write(subscribers[subscriberid] & ", ")
      stdout.write("concerning ")
      for topic in message.topics.items(): stdout.write(topics[topic.int] & " & ")
      stdout.writeLine("\b\b: " & message.data)
    stdout.flushFile()
    if subscriber > -1: echo "--"
  except: discard # write IOError
 
 proc onPull(subscriber: Subscriber, expiredtopics: openArray[Topic], messages: openArray[ptr SuberMessage[string]]) =
  {.gcsafe.}:
    if expiredtopics.len > 0: echo "Sorry, not stocking that old news anymore."
    else: writeMessages(int(subscriber), messages)

proc redeliver() =
  {.gcsafe.}:
    while true:
      if stop: break else: (sleep(2000+rand(3000)) ; if stop: break)
      let since = getMonoTime() - initDuration(milliseconds = 500 + rand(1500))
      echo "" ; echo "--" ; echo("Chas requests Nim-related news since timestamp @", since)
      let subscriber = subscribers.find("Chas")
      bus.pull(subscriber, toIntSet([topics.find("Nim")]), since)

proc onDeliver(messages: openArray[ptr SuberMessage[string]]) {.gcsafe, raises:[].} =
  {.gcsafe.}: writeMessages(-1, messages)

bus.setDeliverCallback(onDeliver)
bus.setPullCallback(onPull)
for i in 0 ..< 4: (for j in 0 .. i: bus.subscribe(i.Subscriber, j.Topic, true))

var delivererthread: Thread[void]
createThread(delivererthread, deliver)
var redelivererthread: Thread[void]
createThread(redelivererthread, redeliver)
var publisherthreads: array[4, Thread[int]]
for i in 0 ..< 4: createThread(publisherthreads[i], generateMessages, i)
joinThreads publisherthreads ; bus.stop()
joinThreads([delivererthread, redelivererthread]) ; echo ""