Reactive Programming - ttulka/programming GitHub Wiki

Reactive Programming Model

Reactive programming is a subset of asynchronous programming and a paradigm where the availability of new information drives the logic forward rather than having control flow driven by a thread-of-execution.

Reactive programming supports decomposing the problem into multiple discrete steps where each can be executed in an asynchronous and non-blocking fashion, and then be composed to produce a workflow—possibly unbounded in its inputs or outputs.

  • Event-driven (events are facts for others to observe) - focuses on addressable event sources.
  • Callback-based and/or Declarative (through functional composition like map, filter, fold etc.) API

Reactive System

Reactive system is an architectural style that allows multiple individual applications to coalesce as a single unit, reacting to its surroundings, while remaining aware of each other—this could manifest as being able to scale up/down, load balancing, and even taking some of these steps proactively.

  • Message-driven (messages have a clear (single) destination) - concentrates on addressable recipients.

Messaging forces us to embrace the reality and constraints of distributed systems (things like partial failures, failure detection, dropped/duplicated/reordered messages, eventual consistency, managing multiple concurrent realities, etc.) and tackle them head on instead of hiding them behind a leaky abstraction—pretending that the network is not there—as has been done too many times in the past (e.g. EJB, RPC, CORBA, and XA).

In a reactive system, especially one which uses reactive programming, both events and messages will be present—as one is a great tool for communication (messages), and another is a great way of representing facts (events).

Actor Programming Model

  • All objects are modeled as independent, computational entities that only respond to the messages received.
  • There is no shared state between actors.
  • Messages are immutable.
  • Messages are passed between actors asynchronously.
  • Each actor has a queue attached where the incoming messages are enqueued.
  • Messages are picked from the queue and processed by the actor, one at a time.
  • An actor can respond to the received message by sending immutable messages to other actors, creating a new set of actors, updating their own state, or designating the computational logic to be used when the next message arrives (behavior change).
  • Communication between the sender and receiver is decoupled and asynchronous, allowing them to execute in different threads.

Actor

  • Actor is an independent, concurrent computational entity that responds to messages.
  • The smallest unit in the grand scheme of things.
  • Can change their state and behavior based on the message passed.

When a mutation is required, a message is sent to the actor, which performs all such changes sequentially, thereby avoiding synchronization problems. A similar model is to allow multiple actors to modify the same state, but only one at a time. A special “semaphore” message is exchanged that tells the receiver that it is safe to modify the state. When finished, the semaphore is sent to another actor.

An actor has no defined interface. An actor has no expectation about which messages other components can understand.

The connection wire between the sender sending a message and the receiver actor receiving the message is called the mailbox.

  • Every actor is attached to exactly one mailbox.

Supervisor hierarchies of actors dedicated to lifecycle management of workers and sophisticated error recovery.

Actor System

  • Container that manages the actor behavior, lifecycle, hierarchy, and configuration among other things.

Example with Akka

public class AkkaActorExample {
  // server code
  static class MemoryActor extends UntypedActor {
    final Map<String, Date> seen = new HashMap<String, Date>();

    public void onReceive(Object messageObject) {
      String message = messageObject.toString(); // simplifying assumption
      if (message.equals("DUMP")) {
        getContext().replySafe(seen.toString());
      } else {
        Date date = new Date();
        seen.put(message.toString(), date);
        getContext().replySafe("'" + message + "' recorded at " + date);
      }
    }
  } 

  public static void main(String[] args) {
    ActorRef remActor = Actors.actorOf(MemoryActor.class).start();
    for (String arg: args) {
      // client code
      Object response = remActor.sendRequestReply(arg);
      System.out.println("Reply received: " + response);
    }
    Object response = remActor.sendRequestReply("DUMP");
    System.out.println("Dump of remembered strings: " + response);
    System.exit(0);
  }
}

Example with vlingo

public class VlingoExampleTest {
  @Test 
  public void testPlayPingPong() {
    final World world = World.start("playground");
    final Pinger pinger = world.actorFor(Definition.has(PingerActor.class, Definition.NoParameters), Pinger.class);
    final Ponger ponger = world.actorFor(Definition.has(PongerActor.class, Definition.NoParameters), Ponger.class);

    pinger.ping(ponger);

    pauseThisThread();
    world.terminate();
  }

  interface Pinger extends Stoppable {
    void ping(Ponger ponger);
  }
  class PingerActor extends Actor implements Pinger {
    private int count = 0;
    private final Pinger self = selfAs(Pinger.class);
  
    public void ping(Ponger ponger) {
      if (++count >= 10) {
        self.stop();
        ponger.stop();
      } else {
        ponger.pong(self);
      }
    }
  }

  interface Ponger extends Stoppable {
    void pong(Pinger pinger);
  }
  class PongerActor extends Actor implements Ponger {
    private final Ponger self = selfAs(Ponger.class);
  
    public void pong(Pinger pinger) {
      pinger.ping(self);
    }
  }
}

References