For those seeing this for the first time, look in the quoted email
history for the email from me showing the outline of RxJava operators...
On Mon, Aug 17, 2015 at 7:20 AM, cogmission (David Ray)
<[email protected] <mailto:[email protected]>> wrote:
Omg! I just found this!
On each of the above operator links (seen in red if you're using
Rich Text or HTML email client), there is a link on the landing
page where you can go to their awesome marble diagrams (which the
have for each operator)...
Except, this one is interactive! So you can move around the
marbles and see the effect in their output (see how the
transformation functions change!)
Check it out! http://rxmarbles.com/#zip
Cheers,
David
On Sun, Aug 16, 2015 at 10:04 PM, cogmission (David Ray)
<[email protected] <mailto:[email protected]>>
wrote:
Hi Matt,
I'm only responding in order to avoid what may possibly be a
missed opportunity, and I don't want you or anyone else who
may see the term "Observable" and mistake it for a simple
publish/subscribe pattern that has been seen before. On the
contrary, this library is extremely powerful, easy on the eyes
and with some subtle usage, can exert extreme power and
flexibility in its usage.
For example:
Say you have a service where you receive a message for the
retrieval of a document on the web. In servicing that request
you need to make an asynchronous query to back-end services,
and then say that service needs to aggregate several formats
from a db, and other streaming sources, so you nest yet
another blocking request to receive those items in a
synchronized fashion and maybe it doesn't end there? Maybe
there is an additional stop along the way where another nested
asynchronous call needs to be made? Now if one of those
connections experiences undue latency or the service is
unavailable for some reason, it is extremely laborious to try
and percolate that exception back up to the original caller.
RxJava handles the bubbling of exceptions from deeply nested
calls back to the original call site.
That's just one example of its use.
Then there are the 100 or so operators that can be performed
on each Observable. Each function returns an Observable that
"emits" one or more items or "streams" of data. And each
Observable can be "operated upon" mathematically or
combined/reduced/transformed etc. And yes, if your Observables
are composed in a "Functional" manner, meaning they don't
mutate or convey state, you are isolated from a lot of
concurrency issues that you would otherwise have to handle.
I'm not saying that a reckless developer can't still get
themselves in trouble - that is true with any breakthrough
library. I'm just saying this is quite an exceptional offering
in the Java language. Other languages might have something
similar - but this is also being used in the C# camp too (and
Ruby, Python, Scala, Clojure, C++, Swift, JRuby, Kotlin etc.)
- Entitled ReactiveX as seen here:
http://reactivex.io/languages.html
This is also different from the Java 8 Streams offering
because it can do both Pull (like Java 8 Streams) and Push
oriented (its major use case) query patterns. As a matter of
fact, I'm sure the authors of the Java 8 release got a lot of
their ideas from this library!
Take a look at some of its operators:
Operators By Category
Creating Observables
Operators that originate new Observables.
* *|Create|*
<http://reactivex.io/documentation/operators/create.html> — create
an Observable from scratch by calling observer methods
programmatically
* *|Defer|*
<http://reactivex.io/documentation/operators/defer.html> —
do not create the Observable until the observer
subscribes, and create a fresh Observable for each observer
* *|Empty|/|Never|/|Throw|*
<http://reactivex.io/documentation/operators/empty-never-throw.html> —
create Observables that have very precise and limited behavior
* *|From|*
<http://reactivex.io/documentation/operators/from.html> —
convert some other object or data structure into an Observable
* *|Interval|*
<http://reactivex.io/documentation/operators/interval.html> —
create an Observable that emits a sequence of integers
spaced by a particular time interval
* *|Just|*
<http://reactivex.io/documentation/operators/just.html> —
convert an object or a set of objects into an Observable
that emits that or those objects
* *|Range|*
<http://reactivex.io/documentation/operators/range.html> —
create an Observable that emits a range of sequential integers
* *|Repeat|*
<http://reactivex.io/documentation/operators/repeat.html> — create
an Observable that emits a particular item or sequence of
items repeatedly
* *|Start|*
<http://reactivex.io/documentation/operators/start.html> —
create an Observable that emits the return value of a function
* *|Timer|*
<http://reactivex.io/documentation/operators/timer.html> —
create an Observable that emits a single item after a
given delay
Transforming Observables
Operators that transform items that are emitted by an Observable.
* *|Buffer|*
<http://reactivex.io/documentation/operators/buffer.html> —
periodically
gather items from an Observable into bundles and emit
these bundles rather than emitting the items one at a time
* *|FlatMap|*
<http://reactivex.io/documentation/operators/flatmap.html> —
transform the items emitted by an Observable into
Observables, then flatten the emissions from those into a
single Observable
* *|GroupBy|*
<http://reactivex.io/documentation/operators/groupby.html> —
divide an Observable into a set of Observables that each
emit a different group of items from the original
Observable, organized by key
* *|Map|*
<http://reactivex.io/documentation/operators/map.html> —
transform the items emitted by an Observable by applying a
function to each item
* *|Scan|*
<http://reactivex.io/documentation/operators/scan.html> —
apply a function to each item emitted by an Observable,
sequentially, and emit each successive value
* *|Window|*
<http://reactivex.io/documentation/operators/window.html> —
periodically
subdivide items from an Observable into Observable windows
and emit these windows rather than emitting the items one
at a time
Filtering Observables
Operators that selectively emit items from a source Observable.
* *|Debounce|*
<http://reactivex.io/documentation/operators/debounce.html> —
only emit an item from an Observable if a particular
timespan has passed without it emitting another item
* *|Distinct|*
<http://reactivex.io/documentation/operators/distinct.html> —
suppress duplicate items emitted by an Observable
* *|ElementAt|*
<http://reactivex.io/documentation/operators/elementat.html> —
emit only item /n/ emitted by an Observable
* *|Filter|*
<http://reactivex.io/documentation/operators/filter.html> — emit
only those items from an Observable that pass a predicate test
* *|First|*
<http://reactivex.io/documentation/operators/first.html> —
emit only the first item, or the first item that meets a
condition, from an Observable
* *|IgnoreElements|*
<http://reactivex.io/documentation/operators/ignoreelements.html> —
do not emit any items from an Observable but mirror its
termination notification
* *|Last|*
<http://reactivex.io/documentation/operators/last.html> —
emit only the last item emitted by an Observable
* *|Sample|*
<http://reactivex.io/documentation/operators/sample.html> — emit
the most recent item emitted by an Observable within
periodic time intervals
* *|Skip|*
<http://reactivex.io/documentation/operators/skip.html> —
suppress the first /n/ items emitted by an Observable
* *|SkipLast|*
<http://reactivex.io/documentation/operators/skiplast.html> —
suppress the last /n/ items emitted by an Observable
* *|Take|*
<http://reactivex.io/documentation/operators/take.html> —
emit only the first /n/ items emitted by an Observable
* *|TakeLast|*
<http://reactivex.io/documentation/operators/takelast.html> —
emit only the last /n/ items emitted by an Observable
Combining Observables
Operators that work with multiple source Observables to create
a single Observable
* *|And|/|Then|/|When|*
<http://reactivex.io/documentation/operators/and-then-when.html> —
combine sets of items emitted by two or more Observables
by means of |Pattern|and |Plan| intermediaries
* *|CombineLatest|*
<http://reactivex.io/documentation/operators/combinelatest.html> —
when an item is emitted by either of two Observables,
combine the latest item emitted by each Observable via a
specified function and emit items based on the results of
this function
* *|Join|*
<http://reactivex.io/documentation/operators/join.html> —
combine items emitted by two Observables whenever an item
from one Observable is emitted during a time window
defined according to an item emitted by the other Observable
* *|Merge|*
<http://reactivex.io/documentation/operators/merge.html> —
combine multiple Observables into one by merging their
emissions
* *|StartWith|*
<http://reactivex.io/documentation/operators/startwith.html> —
emit a specified sequence of items before beginning to
emit the items from the source Observable
* *|Switch|*
<http://reactivex.io/documentation/operators/switch.html> — convert
an Observable that emits Observables into a single
Observable that emits the items emitted by the
most-recently-emitted of those Observables
* *|Zip|*
<http://reactivex.io/documentation/operators/zip.html> —
combine the emissions of multiple Observables together via
a specified function and emit single items for each
combination based on the results of this function
Error Handling Operators
Operators that help to recover from error notifications from
an Observable
* *|Catch|*
<http://reactivex.io/documentation/operators/catch.html> —
recover from an |onError| notification by continuing the
sequence without error
* *|Retry|*
<http://reactivex.io/documentation/operators/retry.html> —
if a source Observable sends an |onError| notification,
resubscribe to it in the hopes that it will complete
without error
Observable Utility Operators
A toolbox of useful Operators for working with Observables
* *|Delay|*
<http://reactivex.io/documentation/operators/delay.html> —
shift the emissions from an Observable forward in time by
a particular amount
* *|Do|*
<http://reactivex.io/documentation/operators/do.html> —
register an action to take upon a variety of Observable
lifecycle events
* *|Materialize|/|Dematerialize|*
<http://reactivex.io/documentation/operators/materialize-dematerialize.html> —
represent both the items emitted and the notifications
sent as emitted items, or reverse this process
* *|ObserveOn|*
<http://reactivex.io/documentation/operators/observeon.html> —
specify the scheduler on which an observer will observe
this Observable
* *|Serialize|*
<http://reactivex.io/documentation/operators/serialize.html> —
force an Observable to make serialized calls and to be
well-behaved
* *|Subscribe|*
<http://reactivex.io/documentation/operators/subscribe.html> —
operate upon the emissions and notifications from an
Observable
* *|SubscribeOn|*
<http://reactivex.io/documentation/operators/subscribeon.html> —
specify the scheduler an Observable should use when it is
subscribed to
* *|TimeInterval|*
<http://reactivex.io/documentation/operators/timeinterval.html> —
convert an Observable that emits items into one that emits
indications of the amount of time elapsed between those
emissions
* *|Timeout|*
<http://reactivex.io/documentation/operators/timeout.html> —
mirror the source Observable, but issue an error
notification if a particular period of time elapses
without any emitted items
* *|Timestamp|*
<http://reactivex.io/documentation/operators/timestamp.html> —
attach a timestamp to each item emitted by an Observable
* *|Using|*
<http://reactivex.io/documentation/operators/using.html> —
create a disposable resource that has the same lifespan as
the Observable
Conditional and Boolean Operators
Operators that evaluate one or more Observables or items
emitted by Observables
* *|All|*
<http://reactivex.io/documentation/operators/all.html> —
determine whether all items emitted by an Observable meet
some criteria
* *|Amb|*
<http://reactivex.io/documentation/operators/amb.html> —
given two or more source Observables, emit all of the
items from only the first of these Observables to emit an item
* *|Contains|*
<http://reactivex.io/documentation/operators/contains.html> —
determine whether an Observable emits a particular item or not
* *|DefaultIfEmpty|*
<http://reactivex.io/documentation/operators/defaultifempty.html> —
emit items from the source Observable, or a default item
if the source Observable emits nothing
* *|SequenceEqual|*
<http://reactivex.io/documentation/operators/sequenceequal.html> —
determine whether two Observables emit the same sequence
of items
* *|SkipUntil|*
<http://reactivex.io/documentation/operators/skipuntil.html> —
discard items emitted by an Observable until a second
Observable emits an item
* *|SkipWhile|*
<http://reactivex.io/documentation/operators/skipwhile.html> —
discard items emitted by an Observable until a specified
condition becomes false
* *|TakeUntil|*
<http://reactivex.io/documentation/operators/takeuntil.html> —
discard items emitted by an Observable after a second
Observable emits an item or terminates
* *|TakeWhile|*
<http://reactivex.io/documentation/operators/takewhile.html> —
discard items emitted by an Observable after a specified
condition becomes false
Mathematical and Aggregate Operators
Operators that operate on the entire sequence of items emitted
by an Observable
* *|Average|*
<http://reactivex.io/documentation/operators/average.html> —
calculates the average of numbers emitted by an Observable
and emits this average
* *|Concat|*
<http://reactivex.io/documentation/operators/concat.html> — emit
the emissions from two or more Observables without
interleaving them
* *|Count|*
<http://reactivex.io/documentation/operators/count.html> —
count the number of items emitted by the source Observable
and emit only this value
* *|Max|*
<http://reactivex.io/documentation/operators/max.html> —
determine, and emit, the maximum-valued item emitted by an
Observable
* *|Min|*
<http://reactivex.io/documentation/operators/min.html> —
determine, and emit, the minimum-valued item emitted by an
Observable
* *|Reduce|*
<http://reactivex.io/documentation/operators/reduce.html> — apply
a function to each item emitted by an Observable,
sequentially, and emit the final value
* *|Sum|*
<http://reactivex.io/documentation/operators/sum.html> —
calculate the sum of numbers emitted by an Observable and
emit this sum
Backpressure Operators
* *backpressure operators*
<http://reactivex.io/documentation/operators/backpressure.html> —
strategies for coping with Observables that produce items
more rapidly than their observers consume them
Connectable Observable Operators
Specialty Observables that have more precisely-controlled
subscription dynamics
* *|Connect|*
<http://reactivex.io/documentation/operators/connect.html> —
instruct a connectable Observable to begin emitting items
to its subscribers
* *|Publish|*
<http://reactivex.io/documentation/operators/publish.html> —
convert an ordinary Observable into a connectable Observable
* *|RefCount|*
<http://reactivex.io/documentation/operators/refcount.html> —
make a Connectable Observable behave like an ordinary
Observable
* *|Replay|*
<http://reactivex.io/documentation/operators/replay.html> — ensure
that all observers see the same sequence of emitted items,
even if they subscribe after the Observable has begun
emitting items
On Sun, Aug 16, 2015 at 8:29 PM, Matthew Lohbihler
<[email protected] <mailto:[email protected]>>
wrote:
Hi David,
It's nice to see that this pattern has been formalized,
but this isn't really anything terribly new. I've been
using observables to do things like rollups and other data
stream analysis for years now, just not with this
particular package.
Also, note that Futures/Promises don't really compete with
Observables. Futures are good for chaining arbitrary
asynchronous events, Observables for processing streams of
homogenous objects. (But is it true that the Java Future
impl is not very useful. I had to develop my own for what
i thought was a very straightforward use case.)
Finally, there is nothing inherently thread safe about
Observables (not that anyone said they were... just
saying...). When using the same Observable instance in a
multi-threaded context, you still have to manage
synchonization on your own if the context doesn't somehow
do it for you.
Regards,
m@
On 8/15/2015 11:56 PM, cogmission (David Ray) wrote:
...and Their Use In HTM.java's Network API (NAPI)
Hi Everybody,
I wanted to share a one-page short but very well
explained tutorial on composing asynchronous workflows
using RxJava.
Due to the availability of "Big Data", Multicore
Processors, Cloud Computing and emerging concepts of
Streaming Workflows; tools like "Hadoop", "Spark" etc.
have come to the forefront as a means of optimizing
concurrency and parallel architectures. Likewise, on a
smaller scale, mastering the tools of Streaming Data
utilization is essential in order to be ready when the
future comes visiting upon us.
Enter RxJava...
As you may or may not know, HTM.java's Network API was
constructed using RxJava Observables which are a way of
composing units of work in such a way where they can be
used asynchronously or non-asynchronously without
blocking. This also allows you to seamlessly "tie in" the
NAPI in your own applications using chains of Observable
items.
Unlike Java "Futures" which only return a single item,
RxJava Observables (which also can use Futures) can
return a "streaming" flow of items in a non-blocking
fashion; this allows one to fully utilize any Threads in
use by letting them be continuously active throughout
their life cycle. Better yet, it is very lightweight has
no outside dependencies and can be used with a multitude
of languages!!
Netflix, uses RxJava and is its original sponsor and
creator. It is a very exciting new concept that is
sweeping the Java developer ranks, and this is one of the
best tutorials on the subject I have found.
http://docs.couchbase.com/developer/java-2.0/observables.html
Enjoy!
David
--
/With kind regards,/
David Ray
Java Solutions Architect
*Cortical.io <http://cortical.io/>*
Sponsor of: HTM.java <https://github.com/numenta/htm.java>
[email protected] <mailto:[email protected]>
http://cortical.io
--
/With kind regards,/
David Ray
Java Solutions Architect
*Cortical.io <http://cortical.io/>*
Sponsor of: HTM.java <https://github.com/numenta/htm.java>
[email protected] <mailto:[email protected]>
http://cortical.io <http://cortical.io/>
--
/With kind regards,/
David Ray
Java Solutions Architect
*Cortical.io <http://cortical.io/>*
Sponsor of: HTM.java <https://github.com/numenta/htm.java>
[email protected] <mailto:[email protected]>
http://cortical.io <http://cortical.io/>
--
/With kind regards,/
David Ray
Java Solutions Architect
*Cortical.io <http://cortical.io/>*
Sponsor of: HTM.java <https://github.com/numenta/htm.java>
[email protected] <mailto:[email protected]>
http://cortical.io <http://cortical.io/>
