Reacto

Reactive Programming for Ruby with some concurrency thrown into the mix!

#Table of Contents

• How to install?
• Why?
• Usage
  • Simple Trackables
    • value
    • error
    • close
    • enumerable
    • interval
    • never
  • Programming Trackable behavior
    • make
    • SharedTrackable
  • Tracking for notifications
    • on
    • track
  • Subscriptions
  • Operations
    • map
    • select
    • inject
    • flat_map
    • ... and even more operations
  • Interacting Trackables
    • merge
    • zip
    • combine
    • concat
    • depend_on
  • Concurency
    • execute_on
    • track_on
    • Executors and factory methods
  • Buffering, delaying and skipping
    • buffer
    • delay
    • throttle
  • Grouping
    • group_by_label
    • chunk
    • operating only on given group
    • flatten_labeled
  • Error handling
    • retrying
    • how to rescue from errors
• Dependencies
• Tested with

How to install?

If you use bundler just add this line to your Gemfile:

gem 'reacto'

Alternatively you can install the gem with gem install reacto or clone this repository and play with it!

Why?

Because it is very cool to be reactive these days! Seriously - to be reactive means that your code is able to react on changes happening in various places right away. For example third party data sources or other parts of the code. It helps writing multi-component apps which could handle failures and still be available.

Of course there are other implementations of reactive programming for ruby:

• RxRuby : Very powerful implementation of RX. Handles concurrency and as a whole has more features than Reacto. So why Reacto? Reacto has simpler interface it is native Ruby lib and is easier to use it. The goal of Reacto is to be alternative to RxRuby in the ruby world. Still the author of Reacto is big fan of RX especially RxJava. He even has a book on the topic using RxJava : Learning Reactive Programming with Java 8
• Frappuccino : Very cool lib, easy to use and simply beautiful. The only drawback - it is a bit limited : no concurrency and small set of operations. But if you don't need more complicated operations it works. The author of Reacto highly recommends it.

Usage

Simple Trackables

The main entry point to the library is the Reacto::Trackable class. It represents something you can track for notifications. Usually a Reacto::Trackable implemenation is pushing notifications to some notification tracker. It depends on the source. We can have some remote streaming service as a source or an asynchronous HTTP request or some process pushing updates to another.

value

Of course the source can be very simple, for example a single value:

  trackable = Reacto::Trackable.value(5)

This value won't be emitted as notification until there is no tracker (listener) attached to trackable - so this Trackable instance is lazy - won't do anything until necessary.

  trackable.on(value: ->(v) { puts v })

  # => 5

This line attaches a notification tracker to the trackable - a lambda that should be called when trackable emits any value. This example is very simple and the trackable emits only one value - 5 when a tracker is attached to it so the lambda will be called and the value will be printed. Shortcuts for Reacto::Trackable.value(val) are Reacto.value(val) and Reacto[val].

error

If we want to emit only an error notification we can do it with Trackable.error. It works the same way as Trackable.value, but the notification is of type error:

  trackable = Reacto::Trackable.error(StandardError.new('Some error!'))

  trackable.on(error: ->(e) { raise e })

Shorcuts for Reacto::Trackable.error(err) are Reacto.error(err) and Reacto[err]. Notice that Reacto[simple_vlue] is like Reacto.value, but Reacto[some_standart_error] is like calling Reacto.error.

close

There is a way to create a Reacto::Trackable emitting only close notification too:

  trackable = Reacto::Trackable.error(StandardError.new('Some error!'))

  trackable.on(close: ->() { p 'closed' })

Shorcuts are Reacto.close and Reacto[:close].

enumerable

Another example is Trackable with source an Enumerable instance:

  trackable = Reacto::Trackable.enumerable([1, 3, 4])

Again we'll have to call #on on it in order to push its values to a tracker. Shorcuts are Reacto.enumerable(enumerable) and Reacto[enumerable].

interval

A neat way to create Trackable emitting the values of some Enumerable on every second, for example is Reacto::Trackable.interval:

  trackable = described_class.interval(0.3)

This one emits the natural numbers on every 0.3 seconds. The second argument can be an Enumerator - limited or unlimited, for example:

  trackable = described_class.interval(2, ('a'..'z').each)

Emits the letters a to z on every two seconds. We can create a custom enumerator and use it. Note that .interval creates Reacto::Trackable which emits in a special thread, so calling #on on it won't block the current thread. Shortcut is Reacto.interval.

never

It is possible that a Trackable which never emits anything is needed. Some operations behave according to Trackable instances returned, so a way to have such a Reacto::Trackable is:

  trackable = Reacto::Trackable.never

Shortcuts for this one are Reacto.never and Reacto[:never]

Programming Trackable behavior

make

A Reacto::Trackable can have custom behavior, defining what and when should be sent:

  trackable = Reacto::Trackable.make do |tracker|
    tracker.on_value('You say yes')
    tracker.on_value('I say no')

    sleep 1
    tracker.on_value('You say stop and I say go go go, oh no')
    tracker.on_close
  end

When a tracker is attached this behavior will become active and the tracker will receive the first two sentences as values, then, after one second the third one and then a closing notification. Shorcut is Reacto.make.

SharedTrackable

Every time a tracker is attached with call to on, this behavior will be executed for the given tracker. If we want to have a shared behavior for all the trackers we can create a Reacto::SharedTrackable instance:

  require 'socket'

  trackable = Reacto::SharedTrackable.make do |subscriber|
    hostname = 'localhost'
    port = 3555

    return unless subscriber.subscribed?

    socket = nil
    begin
      socket = TCPSocket.open(hostname, port)

      while line = socket.gets
        break unless subscriber.subscribed?

        subscriber.on_value(line)
      end

      subscriber.on_close if subscriber.subscribed?
    rescue StandardError => error
      subscriber.on_error(error) if subscriber.subscribed?
    ensure
      socket.close unless socket.nil?
    end

  end

  trackable.on(value: -> (v) { puts v })
  trackable.on do |v|
    puts v
  end

# The above calls of `on` are identical. And the two will the same data.
# Nothing happens on calling `on` though, the `trackable` has to be activated:

  trackable.activate!

Tracking for notifications

on

The easiest way to listen a Reacto::Trackable is to call #on on it:

  consumer = -> (value) do
    # Consume the incoming value
  end

  trackable.on(value: consumer)

Calling it like that will trigger the behavior of trackable and all the values it emits, will be passed to the consumer. A block can be passed to #on and it will have the same effect:

  trackable.on do |value|
    # Consume the incoming value
  end

If we want to listen for errors we can call #on like that:

  error_consumer = -> (error) do
    # Consume the incoming error
  end

  trackable.on(error: error_consumer)

Only one error can be emitted by a Trackable for subscription and that will close the Reacto::Trackable. If there is no error, the normal closing notification should be emitted. We can fetch it like this:

  to_be_called_on_close = -> () do
    # Finalize?
  end

  trackable.on(close: to_be_called_on_close)

track

Under the hood #on creates a new Reacto::Tracker instance with the right methods. If we want to create our own tracker, we can always call #track on the trackable with the given instance:

  consumer = -> (value) do
    # Consume the incoming value
  end
  error_consumer = -> (error) do
    # Consume the incoming error
  end

  trackable.track(Reacto::Trackable.new(
    value: consumer, error: error_consumer, close: -> () { p 'Closing!' }
  ))

All of these keyword parameters have default values - for example if we don't pass a value: action, a no-action will be used, doing nothing with the value, the sa