Or an adventure on reading data reactively from Cassandra.<\/p>\n
Let’s first try to define what reactive means from programming point of view.<\/p>\n
Functional reactive programming <\/a>is programming paradigm for reactive programming using the building blocks of functional programming.<\/p>\n Functional programming<\/a> is a programming paradigm, a style of building the structure and the elements of computer programs, that treats computation, as the evaluation of mathematical functions thats avoids state and mutable data. Functional programming emphasises functions that produce results that depend only on \u00a0their inputs and not on program state.<\/p>\n How can we do functional programming in Java? Java is Object Oriented Programming language where mutable state is present everywhere.<\/p>\n Any java developer around the world have used any of the interfaces: <\/p>\n java.lang.Runnable, java.util.Comparator, java.util.concurrent.Callable or java.awt.event.ActionListener. All of this interfaces are having only single method declared. These interfaces are known as Single Abstract Methods or SAM. A popular way as these are used is by creating Anonymous inner classes.<\/p>\n Functional Programming in Java is hard since function is not included in the language specification. It will become simpler in Java 8 with introduction of \u201clambda\u2019s\u201d. But how can we do functional programming in Java?<\/p>\n Let\u2019s see a simple example.<\/p>\n Reactive programming<\/a> is a programming paradigm oriented around data flows and the propagation of changes. For example, in imperative programming setting, a := b+c, would mean that a is being assigned the result of \u00a0b +c in the instant the expression is evaluated. Later values of b or c can be changed without effect on a. In reactive programming, the value of a will be automatically updated based on the new values. So, we should have a good understanding of what Functional Reactive Programming is, so let’s go and build a prototype…<\/p>\n Cassandra is one of the NoSql storage out there is quite popular.<\/p>\n Let’s imagine that we have to build an Avatar service. This service will store avatars meta information and the content directly in cassandra.<\/p>\n The java driver that we are using is providing us support to query cassandra asynchronous, through the executeAsync()<\/a> method. The invocation of this method will return a Future. As we all know java Futures are block-able and could not be composed.<\/p>\n Ok, so we have async support but we still need a way to be able to read it reactively…<\/p>\n Netflix built and later open sourced the RxJava<\/a> library that is providing Functional Reactive Programming for Java (plus other JVM languages).<\/p>\n Functional reactive offers efficient execution and composition by providing a collection of operators capable of filtering, selecting, transforming, combining and composing Observable’s.<\/p>\n The Observable data type can be thought of as a “push” equivalent to Iterable<\/a> which is “pull”. With an Iterable, the consumer pulls values from the producer and the thread blocks until those values arrive. By contrast with the Observable type, the producer pushes values to the consumer whenever values are available. This approach is more flexible, because values can arrive synchronously or asynchronously.<\/p>\n The Observable<\/a> type adds two missing semantics to the Gang of Four’s Observer<\/a> pattern, which are available in the Iterable type:<\/p>\n With these two simple additions, we have unified the Iterable and Observable types. The only difference between them is the direction in which the data flows. This is very important because now any operation we perform on an Iterable, can also be performed on an Observable.<\/p>\n Let’s see how we can combine the RxJava and Cassandra async query execution to build an Observable. executeAsync() <\/a>method is returning a Guava Listenable Future<\/a>. Adding a callback on this future allows us to properly implement the Observer<\/a> interface.<\/p>\n A simple service could be implemented as following:<\/p>\n Assuming that the query is heavy, we are providing a separate execution context where the callback will be executed.<\/p>\n With these two classes we have an Avatar service that could be started but it will not do any thing. It will start fetching the data from Cassandra only when there will be at least one subscriber. A complete example could be found at Reactive Cassandra<\/a>. Or an adventure on reading data reactively from Cassandra. Overview Let’s first try to define what reactive means from programming point of view. Functional reactive programming is programming paradigm for reactive programming using the building blocks of functional programming. Functional programming is a programming paradigm, a style of building the structure and the elements of …<\/p>\n","protected":false},"author":82,"featured_media":53,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[8],"tags":[111,113],"class_list":["post-20246","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-enterprise-java","tag-apache-cassandra","tag-nosql"],"yoast_head":"\npublic class RunnableTest {\r\n public static void main(Sting[] args){\r\n new Thread(new Runnable(){\r\n @Override\r\n public void run(){\r\n System.out.println(\"A new thread is running ...\");\r\n }\r\n }).start();\r\n }\r\n}<\/pre>\n@FunctionalInterface\r\npublic interface Worker {\r\n public void doWork();\r\n}\r\npublic class FunctionalWorker {\r\n public static void main(String[] args){\r\n \/\/ anonymous inner class way\r\n execute( new Worker(){\r\n @Override\r\n public void doWork() {\r\n System.out.println (\"working ...\");\r\n }\r\n });\r\n \/\/ lambda's way\r\n execute(() -> System.out.println(\"working in lambda's way ...\"));\r\n }\r\n\r\n public static void execute(Worker worker){\r\n worker.doWork();\r\n }\r\n}<\/pre>\nReading data\u00a0reactively\u00a0from Cassandra<\/h2>\n
\n
\n<\/code><\/p>\npackage net.devsprint.reactive.cassandra;\r\n\r\nimport java.util.concurrent.ExecutorService;\r\n\r\nimport rx.Observable;\r\nimport rx.Observer;\r\nimport rx.Subscription;\r\nimport rx.subscriptions.Subscriptions;\r\n\r\nimport com.datastax.driver.core.ResultSet;\r\nimport com.datastax.driver.core.Session;\r\nimport com.google.common.util.concurrent.FutureCallback;\r\nimport com.google.common.util.concurrent.Futures;\r\n\r\n\/**\r\n * Wraps an async execution of Datastax Java driver into an observable.\r\n * \r\n *\/\r\npublic class ObservableCassandra {\r\n\r\n public static Observable executeAsync(final Session execution,\r\n final String query, final ExecutorService executorService) {\r\n return Observable.create(new Observable.OnSubscribeFunc() {\r\n\r\n @Override\r\n public Subscription onSubscribe(final Observer observer) {\r\n try {\r\n Futures.addCallback(execution.executeAsync(query),\r\n new FutureCallback() {\r\n\r\n @Override\r\n public void onSuccess(ResultSet result) {\r\n observer.onNext(result);\r\n observer.onCompleted();\r\n }\r\n\r\n @Override\r\n public void onFailure(Throwable t) {\r\n observer.onError(t);\r\n }\r\n }, executorService);\r\n } catch (Throwable e) {\r\n \/\/ If any Throwable can be thrown from\r\n \/\/ executeAsync\r\n observer.onError(e);\r\n }\r\n return Subscriptions.empty();\r\n }\r\n });\r\n }\r\n\r\n}<\/pre>\npackage net.devsprint.reactive.cassandra;\r\n\r\nimport java.util.concurrent.ExecutorService;\r\nimport java.util.concurrent.Executors;\r\n\r\nimport rx.Observable;\r\n\r\nimport com.datastax.driver.core.ResultSet;\r\nimport com.datastax.driver.core.Session;\r\n\r\npublic class AvatarService {\r\n private static final String QUERY = \"select * avatars\";\r\n private static final ExecutorService executorService = Executors\r\n .newFixedThreadPool(Runtime.getRuntime().availableProcessors());\r\n\r\n private final Session session;\r\n\r\n public AvatarService(Session session) {\r\n this.session = session;\r\n }\r\n\r\n Observable getAvatars() {\r\n return ObservableCassandra\r\n .executeAsync(session, QUERY, executorService);\r\n }\r\n\r\n}<\/pre>\n
\n <\/p>\n