agourlay/omnibus

An HTTP-friendly persistent message bus.

omnibus's Issues

Test REST API with ScalaCheck

Test spray interface using topic name randomly generate via Scalacheck

Allow configuration

The standalone process should be configurable with a file in /conf/application.conf

• port
• data rentention time

etc

see : https://github.com/agourlay/omnibus/blob/master/src/main/resources/application.conf#L1

Grafana dashboard

Provide a Grafana dashboard to display metrics from Graphite

http://grafana.org/
https://www.youtube.com/watch?v=OUvJamHeMpw

If you publish a message at the "/animals" level, all subtopics will receive it as well.

Interesting design decision .. I would have said that if one listens for /topics/animals one gets all animal related messages (incl. furry) but if one listens on /topics/animals/furry then one just gets those and not messages from /topics/animals

Thoughts?

Restructure application for better testability

Follow the great example here

http://www.cakesolutions.net/teamblogs/2013/08/02/akka-and-spray/

Re-work persistence

Omnibus persists locally a journal log of events on the file system.

Currently it uses akka-persistence-leveldb which has performance issues :
akka/akka#15585

ideas :

• do not use akka-persistence as it is not made for durable queue :
  https://groups.google.com/forum/#!topic/akka-user/5fRv72GFE4E
• use unix sendfile() :
  https://www.ibm.com/developerworks/linux/library/j-zerocopy/
• manage R/W for files manually :
  https://github.com/drexin/akka-io-file

Integrate performance test to build.

Use gatling-sbt plugin when ready https://github.com/gatling/gatling-sbt

Websocket

Would be awesome to pub/sub via websocket.
Wait for websocket to be implemented in akka-http

Minimal administration interface

provide a basic administration interface to

• list topics
• display http system stats
• display topics stats
• protected by basic authentication

Investigate one actor per-request pattern

Currently the ´ask pattern´ is used often.

Investigate the pattern one actor per-request.
http://techblog.net-a-porter.com/2013/12/ask-tell-and-per-request-actors/

Try to remove all usage of Future.

Create and delete topic via admin interface

Manipulate the rest API via the admin interface for lazy people.

• create topic and sub topic
• delete topic

Hateoas approach

Right now the GET method on a topic is used to subscribe directly to the topic.

I believe we should offer a level of indirection to expose the topic in a more REST fashion way.

GET /topics/animals/furry "should" reply something like

{
  "name" : "animals/furry",
  "creationDate" : "aTimestamp",
   "_links": {
        "subscribe": { "href": "/stream/topics/animals/furry" },
        "stats":  { "href": "/stats/topics/animals/furry" },
        "parent": { "href": "/topic/animals/" },
        "children": [{ "href": "/topic/animals/furry/cats",
                          "href": "/topic/animals/furry/fox" }]
    }
}

That way the topic tree can be traversed in a clear REST way and subscribing becomes more clear with the action url.

Any suggestion/feedback on this?

edit : fix title

Application metrics

Redesign application metrics using Codahale's Metrics http://metrics.codahale.com/manual/

Archived stats. should not be saved in Cassandra with the events but pushed outside to a reporter (Graphite for instance).

Managing stats manually will become a burden in a distributed environment where stats need to be aggregated.

Omnibus will only provide the current stats in the webadmin.

Handle HTTP header "Last-Event-ID"

"Last-Event-ID" can be directly used with "since-id" reactive mode,

Support binary payload

optional binary payload per connection.

Use sbt-native-packager

Build omnibus standalone with https://github.com/sbt/sbt-native-packager

Topic's name should not be represented as String in internal API

Using String for topic's name introduce ugly code and potential bugs due to lack of input validation.

Create a case class TopicPath that would contain a list of segments and a toString method to produce the printed version.

The topic name coming from spray interface should be validated and serialized into TopicPath as soon as possible.

All actors should then use TopicPath in their API and state.

The subTopics creation should propagate the initial post

When a topic is created, the message in POST data is received by subscribers whereas on subtopic this message is not received.

Admin API for to manage subscribers

Design a domain object for subscription (topic, IP, id etc) and expose API.

get -> /admin/subscribers : get collection of all subscribers
delete -> /admin/subscribers/{sub_id} : shutdown connection

Basic REST compliance

make sure that responses contain the proper status code and clean error messages

• create topic with 201/202 and proper location header
• delete topic 404 if topic not found
• publish to topic 404 if topic not found
• get topic 404 if topic not found
• get stats topic 404 if topic not found
• get stream topic 404 if topic not found

Support external persistence solution

Omnibus should support an external akka-persistence plugin to persist data

• cassandra https://github.com/krasserm/akka-persistence-cassandra

pagination of /topics & /subscribers

those collections can become pretty big

Scala 2.11

Migrate to Scala 2.11

The stats response doesn't provide the global information of topic

The stat should provide the complete name of the topic; not only the current node topic name.
should be discussed :

1. should add a new originEvent member : original event arrive on this topic; to be add on all stat
2. should add a new topicPath member : complete path to the topic; can be store on a extern member of the list of stats as it will the same for all stats of this topic.

In the current stat response for a /stats/topics/omnibus/bus1
we got a topicName "bus1" but it doesn't help for aggregate stat if there is another /topics/personalbus/bus1; we can't say if it we need to add an entry for the first or the second.

Deploy bus in cluster

Some ideas :

• frontend-nodes taking care of http tasks
• backend-nodes with topics
• the topic tree should be sharded and replicated properly
• akka-cluster-sharding
• Distrib. Pub/Sub extension http://doc.akka.io/docs/akka/snapshot/contrib/distributed-pub-sub.html
• the actor structure should not reflect the topic hierarchy as it introduces bottlenecks
• akka-data-replication to create a distr. file system

Support MQTT

Omnibus acts as a MQTT broker

can't start the app

[info] Running omnibus.Boot
2014-11-19 10:06:08,847 INFO akka.event.slf4j.Slf4jLogger - Slf4jLogger started
Uncaught error from thread [omnibus-akka.persistence.dispatchers.default-plugin-dispatcher-9] shutting down JVM since 'akka.jvm-exit-on-fatal-error' is enabled for ActorSystem[omnibus]
java.lang.UnsatisfiedLinkError: org.fusesource.leveldbjni.internal.NativeOptions.init()V
at org.fusesource.leveldbjni.internal.NativeOptions.init(Native Method)
at org.fusesource.leveldbjni.internal.NativeOptions.(NativeOptions.java:54)
at org.fusesource.leveldbjni.JniDBFactory$OptionsResourceHolder.init(JniDBFactory.java:98)
at org.fusesource.leveldbjni.JniDBFactory.open(JniDBFactory.java:167)
at akka.persistence.journal.leveldb.LeveldbStore$class.preStart(LeveldbStore.scala:112)
at akka.persistence.journal.leveldb.LeveldbJournal.preStart(LeveldbJournal.scala:20)
at akka.actor.Actor$class.aroundPreStart(Actor.scala:470)
at akka.persistence.journal.leveldb.LeveldbJournal.aroundPreStart(LeveldbJournal.scala:20)
at akka.actor.ActorCell.create(ActorCell.scala:580)
at akka.actor.ActorCell.invokeAll$1(ActorCell.scala:456)
at akka.actor.ActorCell.systemInvoke(ActorCell.scala:478)
at akka.dispatch.Mailbox.processAllSystemMessages(Mailbox.scala:279)
at akka.dispatch.Mailbox.run(Mailbox.scala:220)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)
at java.lang.Thread.run(Thread.java:722)