reiseburo
/
beetle
mirror of https://github.com/reiseburo/beetle
1
0
Fork 0
Code Issues Releases Wiki Activity

Sketching out some more example prototypes

This commit is contained in:
R. Tyler Croy 2015-09-03 15:51:53 -07:00
parent 450d0bd0c9
commit 36654bd721
No known key found for this signature in database
GPG Key ID: 1426C7DC3F51E16F
1 changed files with 34 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
README.adoc
View File

@ -44,24 +44,52 @@ Subscribers, e.g.g
.Consumer.java .Consumer.java
[source, java] [source, java]
---- ----
/* Using Java 8 Lambda syntax for conciseness' sake */ /*
LocateBrokers.zookeeper("localhost:2181") * Prototype code showing how a typical end-user might use Beetle
.map(broker -> TopicSubscription("some-topic")) */
Brokers.fromZookeeper("localhost:2181")
/* assuming a custom subscribe() operator exists in Beetle */
.subscribe("some-topic")
/* assuming a custom consume() operator exists in Beetle */ /* assuming a custom consume() operator exists in Beetle */
.consume(message -> doSomethingWithMessage(message)) .consume(message -> doSomethingWithMessage(message))
.map(message -> message.commitOffset()); .map(message -> message.commitOffset());
---- ----
.ZookeeperlessConsumer.java .ZookeeperlessConsumer.java
[source, java] [source, java]
---- ----
Broker.just('localhost:6667') /*
.map(broker -> TopicSubscription("some-topic") * Assuming we already know which broker we want to talk to and
* we don't care at all about leader changes or committing offsets
*/
long startOffset = 0;
Brokers.just('localhost:6667')
.subscribe("some-topic", startOffset)
.consume(m -> doSomethingWithMessage(m)) .consume(m -> doSomethingWithMessage(m))
.map(m -> m.commitOffset());
---- ----
.LowLevelBeetle.java
----
/*
* The following is a prototype of some ideas for how the above examples
* might be implemented internally
*/
CuratorFramework cf = CuratorFrameworkFactory.newClient("localhost:2181");
BrokersObserver.observe(cf)
.subscribe(brokers -> TopicsObserver.observe(cf, brokers))
.subscribe(partitions -> ConsumerObserver.observe(cf, partitions))
.map(message ->
doCustomBehaviorWith(message))
.map(message -> m.commitOffsetTo(cf));
----
== Similar Projects == Similar Projects
. link:https://github.com/cjdev/kafka-rx[kafka-rx]: Scala-based client which . link:https://github.com/cjdev/kafka-rx[kafka-rx]: Scala-based client which