Akka/Scala example source code file (JavaReadBackPressure.java)

This example Akka source code file (JavaReadBackPressure.java) is included in my "Source Code Warehouse" project. The intent of this project is to help you more easily find Akka and Scala source code examples by using tags.

All credit for the original source code belongs to akka.io; I'm just trying to make examples easier to find. (For my Scala work, see my Scala examples and tutorials.)

Akka tags/keywords

ack, actor, actorref, akka, bytestring, exception, inetsocketaddress, io, javareadbackpressure, listener, override, pullecho, untypedactor, util

The JavaReadBackPressure.java Akka example source code

package docs.io;

import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.io.Inet;
import akka.io.Tcp;
import akka.io.TcpMessage;
import akka.util.ByteString;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;

/**
 * Copyright (C) 2009-2014 Typesafe Inc. <http://www.typesafe.com>
 */
public class JavaReadBackPressure {

    static public class Listener extends UntypedActor {
        ActorRef tcp;
        ActorRef listener;

        @Override
        //#pull-accepting
        public void onReceive(Object message) throws Exception {
            if (message instanceof Tcp.Bound) {
                listener = getSender();
                // Accept connections one by one
                listener.tell(TcpMessage.resumeAccepting(1), getSelf());
            } else if (message instanceof Tcp.Connected) {
                ActorRef handler = getContext().actorOf(Props.create(PullEcho.class, getSender()));
                getSender().tell(TcpMessage.register(handler), getSelf());
                // Resume accepting connections
                listener.tell(TcpMessage.resumeAccepting(1), getSelf());
            }
        }
        //#pull-accepting

        @Override
        public void preStart() throws Exception {
            //#pull-mode-bind
            tcp = Tcp.get(getContext().system()).manager();
            final List<Inet.SocketOption> options = new ArrayList<Inet.SocketOption>();
            tcp.tell(
               TcpMessage.bind(getSelf(), new InetSocketAddress("localhost", 0), 100, options, true),
               getSelf()
            );
            //#pull-mode-bind
        }

        private void demonstrateConnect() {
            //#pull-mode-connect
            final List<Inet.SocketOption> options = new ArrayList<Inet.SocketOption>();
            tcp.tell(
               TcpMessage.connect(new InetSocketAddress("localhost", 3000), null, options, null, true),
               getSelf()
            );
            //#pull-mode-connect
        }
    }

    static public class Ack implements Tcp.Event {
    }

    static public class PullEcho extends UntypedActor {
        final ActorRef connection;

        public PullEcho(ActorRef connection) {
            this.connection = connection;
        }

        //#pull-reading-echo
        @Override
        public void preStart() throws Exception {
            connection.tell(TcpMessage.resumeReading(), getSelf());
        }

        @Override
        public void onReceive(Object message) throws Exception {
            if (message instanceof Tcp.Received) {
                ByteString data = ((Tcp.Received) message).data();
                connection.tell(TcpMessage.write(data, new Ack()), getSelf());
            } else if (message instanceof Ack) {
                connection.tell(TcpMessage.resumeReading(), getSelf());
            }
        }
        //#pull-reading-echo


    }

}