implement ResumeWriting, see #3200

also included:
- a complete rewrite of the TCP docs based on real/tested/working code
  samples
- an EchoServer implementation which handles all the edge cases,
  available in Java & Scala
- renamed StopReading to SuspendReading to match up with ResumeReading
- addition of Inbox.watch()
- Inbox RST docs for Java(!) and Scala

not included:
- ScalaDoc / JavaDoc for all IO stuff

akka-docs/rst/java/code/docs/actor/InboxDocTest.java

@@ -0,0 +1,64 @@
+/**
+ * Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package docs.actor;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import scala.concurrent.duration.Duration;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Inbox;
+import akka.actor.PoisonPill;
+import akka.actor.Terminated;
+import akka.testkit.AkkaSpec;
+import akka.testkit.JavaTestKit;
+
+public class InboxDocTest {
+
+  private static ActorSystem system;
+
+  @BeforeClass
+  public static void beforeAll() {
+    system = ActorSystem.create("MySystem", AkkaSpec.testConf());
+  }
+
+  @AfterClass
+  public static void afterAll() {
+    system.shutdown();
+    system = null;
+  }
+  
+  @Test
+  public void demonstrateInbox() {
+    final JavaTestKit probe = new JavaTestKit(system);
+    final ActorRef target = probe.getRef();
+    //#inbox
+    final Inbox inbox = Inbox.create(system);
+    inbox.send(target, "hello");
+    //#inbox
+    probe.expectMsgEquals("hello");
+    probe.send(probe.getLastSender(), "world");
+    //#inbox
+    assert inbox.receive(Duration.create(1, TimeUnit.SECONDS)).equals("world");
+    //#inbox
+  }
+  
+  @Test
+  public void demonstrateWatch() {
+    final JavaTestKit probe = new JavaTestKit(system);
+    final ActorRef target = probe.getRef();
+    //#watch
+    final Inbox inbox = Inbox.create(system);
+    inbox.watch(target);
+    target.tell(PoisonPill.getInstance(), null);
+    assert inbox.receive(Duration.create(1, TimeUnit.SECONDS)) instanceof Terminated;
+    //#watch
+  }
+
+}

akka-docs/rst/java/code/docs/io/japi/EchoHandler.java

@@ -0,0 +1,229 @@
+/**
+ * Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package docs.io.japi;
+
+import java.net.InetSocketAddress;
+import java.util.LinkedList;
+import java.util.Queue;
+
+import akka.actor.ActorRef;
+import akka.actor.UntypedActor;
+import akka.event.Logging;
+import akka.event.LoggingAdapter;
+import akka.io.Tcp.CommandFailed;
+import akka.io.Tcp.ConnectionClosed;
+import akka.io.Tcp.Received;
+import akka.io.Tcp.Write;
+import akka.io.Tcp.WritingResumed;
+import akka.io.TcpMessage;
+import akka.japi.Procedure;
+import akka.util.ByteString;
+
+//#echo-handler
+public class EchoHandler extends UntypedActor {
+
+  final LoggingAdapter log = Logging
+      .getLogger(getContext().system(), getSelf());
+
+  final ActorRef connection;
+  final InetSocketAddress remote;
+
+  public static final long MAX_STORED = 100000000;
+  public static final long HIGH_WATERMARK = MAX_STORED * 5 / 10;
+  public static final long LOW_WATERMARK = MAX_STORED * 2 / 10;
+
+  public EchoHandler(ActorRef connection, InetSocketAddress remote) {
+    this.connection = connection;
+    this.remote = remote;
+
+    // sign death pact: this actor stops when the connection is closed
+    getContext().watch(connection);
+
+    // start out in optimistic write-through mode
+    getContext().become(writing);
+  }
+
+  private final Procedure<Object> writing = new Procedure<Object>() {
+    @Override
+    public void apply(Object msg) throws Exception {
+      if (msg instanceof Received) {
+        final ByteString data = ((Received) msg).data();
+        connection.tell(TcpMessage.write(data, currentOffset()), getSelf());
+        buffer(data);
+
+      } else if (msg instanceof Integer) {
+        acknowledge((Integer) msg);
+
+      } else if (msg instanceof CommandFailed) {
+        final Write w = (Write) ((CommandFailed) msg).cmd();
+        connection.tell(TcpMessage.resumeWriting(), getSelf());
+        getContext().become(buffering((Integer) w.ack()));
+
+      } else if (msg instanceof ConnectionClosed) {
+        final ConnectionClosed cl = (ConnectionClosed) msg;
+        if (cl.isPeerClosed()) {
+          if (storage.isEmpty()) {
+            getContext().stop(getSelf());
+          } else {
+            getContext().become(closing);
+          }
+        }
+      }
+    }
+  };
+
+  //#buffering
+  protected Procedure<Object> buffering(final int nack) {
+    return new Procedure<Object>() {
+
+      private int toAck = 10;
+      private boolean peerClosed = false;
+
+      @Override
+      public void apply(Object msg) throws Exception {
+        if (msg instanceof Received) {
+          buffer(((Received) msg).data());
+
+        } else if (msg instanceof WritingResumed) {
+          writeFirst();
+
+        } else if (msg instanceof ConnectionClosed) {
+          if (((ConnectionClosed) msg).isPeerClosed())
+            peerClosed = true;
+          else
+            getContext().stop(getSelf());
+
+        } else if (msg instanceof Integer) {
+          final int ack = (Integer) msg;
+          acknowledge(ack);
+
+          if (ack >= nack) {
+            // otherwise it was the ack of the last successful write
+
+            if (storage.isEmpty()) {
+              if (peerClosed)
+                getContext().stop(getSelf());
+              else
+                getContext().become(writing);
+
+            } else {
+              if (toAck > 0) {
+                // stay in ACK-based mode for a short while
+                writeFirst();
+                --toAck;
+              } else {
+                // then return to NACK-based again
+                writeAll();
+                if (peerClosed)
+                  getContext().become(closing);
+                else
+                  getContext().become(writing);
+              }
+            }
+          }
+        }
+      }
+    };
+  }
+  //#buffering
+
+  //#closing
+  protected Procedure<Object> closing = new Procedure<Object>() {
+    @Override
+    public void apply(Object msg) throws Exception {
+      if (msg instanceof CommandFailed) {
+        // the command can only have been a Write
+        connection.tell(TcpMessage.resumeWriting(), getSelf());
+        getContext().become(closeResend, false);
+      } else if (msg instanceof Integer) {
+        acknowledge((Integer) msg);
+        if (storage.isEmpty())
+          getContext().stop(getSelf());
+      }
+    }
+  };
+
+  protected Procedure<Object> closeResend = new Procedure<Object>() {
+    @Override
+    public void apply(Object msg) throws Exception {
+      if (msg instanceof WritingResumed) {
+        writeAll();
+        getContext().unbecome();
+      } else if (msg instanceof Integer) {
+        acknowledge((Integer) msg);
+      }
+    }
+  };
+  //#closing
+
+  //#storage-omitted
+  @Override
+  public void onReceive(Object msg) throws Exception {
+    // this method is not used due to become()
+  }
+
+  @Override
+  public void postStop() {
+    log.info("transferred {} bytes from/to [{}]", transferred, remote);
+  }
+
+  private long transferred;
+  private int storageOffset = 0;
+  private long stored = 0;
+  private Queue<ByteString> storage = new LinkedList<ByteString>();
+
+  private boolean suspended = false;
+
+  //#helpers
+  protected void buffer(ByteString data) {
+    storage.add(data);
+    stored += data.size();
+
+    if (stored > MAX_STORED) {
+      log.warning("drop connection to [{}] (buffer overrun)", remote);
+      getContext().stop(getSelf());
+
+    } else if (stored > HIGH_WATERMARK) {
+      log.debug("suspending reading at {}", currentOffset());
+      connection.tell(TcpMessage.suspendReading(), getSelf());
+      suspended = true;
+    }
+  }
+
+  protected void acknowledge(int ack) {
+    assert ack == storageOffset;
+    assert !storage.isEmpty();
+
+    final ByteString acked = storage.remove();
+    stored -= acked.size();
+    transferred += acked.size();
+    storageOffset += 1;
+
+    if (suspended && stored < LOW_WATERMARK) {
+      log.debug("resuming reading");
+      connection.tell(TcpMessage.resumeReading(), getSelf());
+      suspended = false;
+    }
+  }
+  //#helpers
+
+  protected int currentOffset() {
+    return storageOffset + storage.size();
+  }
+
+  protected void writeAll() {
+    int i = 0;
+    for (ByteString data : storage) {
+      connection.tell(TcpMessage.write(data, storageOffset + i++), getSelf());
+    }
+  }
+
+  protected void writeFirst() {
+    connection.tell(TcpMessage.write(storage.peek(), storageOffset), getSelf());
+  }
+
+  //#storage-omitted
+}
+//#echo-handler

akka-docs/rst/java/code/docs/io/japi/EchoManager.java

@@ -0,0 +1,81 @@
+/**
+ * Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package docs.io.japi;
+
+import java.net.InetSocketAddress;
+
+import akka.actor.ActorRef;
+import akka.actor.Props;
+import akka.actor.SupervisorStrategy;
+import akka.actor.UntypedActor;
+import akka.event.Logging;
+import akka.event.LoggingAdapter;
+import akka.io.Tcp;
+import akka.io.Tcp.Bind;
+import akka.io.Tcp.Bound;
+import akka.io.Tcp.CommandFailed;
+import akka.io.Tcp.Connected;
+import akka.io.TcpMessage;
+
+public class EchoManager extends UntypedActor {
+
+  final LoggingAdapter log = Logging
+      .getLogger(getContext().system(), getSelf());
+
+  final Class<?> handlerClass;
+
+  public EchoManager(Class<?> handlerClass) {
+    this.handlerClass = handlerClass;
+  }
+
+  @Override
+  public SupervisorStrategy supervisorStrategy() {
+    return SupervisorStrategy.stoppingStrategy();
+  }
+
+  @Override
+  public void preStart() throws Exception {
+    //#manager
+    final ActorRef tcpManager = Tcp.get(getContext().system()).manager();
+    //#manager
+    tcpManager.tell(
+        TcpMessage.bind(getSelf(), new InetSocketAddress("localhost", 0), 100),
+        getSelf());
+  }
+
+  @Override
+  public void postRestart(Throwable arg0) throws Exception {
+    // do not restart
+    getContext().stop(getSelf());
+  }
+
+  @Override
+  public void onReceive(Object msg) throws Exception {
+    if (msg instanceof Bound) {
+      log.info("listening on [{}]", ((Bound) msg).localAddress());
+    } else if (msg instanceof Tcp.CommandFailed) {
+      final CommandFailed failed = (CommandFailed) msg;
+      if (failed.cmd() instanceof Bind) {
+        log.warning("cannot bind to [{}]", ((Bind) failed.cmd()).endpoint());
+        getContext().stop(getSelf());
+      } else {
+        log.warning("unknown command failed [{}]", failed.cmd());
+      }
+    } else
+    if (msg instanceof Connected) {
+      final Connected conn = (Connected) msg;
+      log.info("received connection from [{}]", conn.remoteAddress());
+      final ActorRef connection = getSender();
+      final ActorRef handler = getContext().actorOf(
+          Props.create(handlerClass, connection, conn.remoteAddress()));
+      //#echo-manager
+      connection.tell(TcpMessage.register(handler,
+          true, // <-- keepOpenOnPeerClosed flag
+          true), getSelf());
+      //#echo-manager
+    }
+  }
+
+}

akka-docs/rst/java/code/docs/io/japi/EchoServer.java

@@ -0,0 +1,35 @@
+/**
+ * Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package docs.io.japi;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+
+public class EchoServer {
+
+  public static void main(String[] args) throws InterruptedException {
+    final Config config = ConfigFactory.parseString("akka.loglevel=DEBUG");
+    final ActorSystem system = ActorSystem.create("EchoServer", config);
+    try {
+      final CountDownLatch latch = new CountDownLatch(1);
+      final ActorRef watcher = system.actorOf(Props.create(Watcher.class, latch), "watcher");
+      final ActorRef nackServer = system.actorOf(Props.create(EchoManager.class, EchoHandler.class), "nack");
+      final ActorRef ackServer = system.actorOf(Props.create(EchoManager.class, SimpleEchoHandler.class), "ack");
+      watcher.tell(nackServer, null);
+      watcher.tell(ackServer, null);
+      latch.await(10, TimeUnit.MINUTES);
+    } finally {
+      system.shutdown();
+    }
+  }
+
+}

akka-docs/rst/java/code/docs/io/japi/IODocTest.java

@@ -0,0 +1,179 @@
+/**
+ * Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package docs.io.japi;
+
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+//#imports
+import java.net.InetSocketAddress;
+import akka.actor.ActorRef;
+import akka.actor.ActorSystem;
+import akka.actor.Props;
+import akka.actor.UntypedActor;
+import akka.io.Tcp;
+import akka.io.Tcp.Bound;
+import akka.io.Tcp.CommandFailed;
+import akka.io.Tcp.Connected;
+import akka.io.Tcp.ConnectionClosed;
+import akka.io.Tcp.Received;
+import akka.io.TcpMessage;
+import akka.japi.Procedure;
+import akka.util.ByteString;
+//#imports
+
+import akka.testkit.JavaTestKit;
+import akka.testkit.AkkaSpec;
+
+public class IODocTest {
+
+  static
+  //#server
+  public class Server extends UntypedActor {
+    
+    final ActorRef manager;
+    
+    public Server(ActorRef manager) {
+      this.manager = manager;
+    }
+
+    @Override
+    public void preStart() throws Exception {
+      final ActorRef tcp = Tcp.get(getContext().system()).manager();
+      tcp.tell(TcpMessage.bind(getSelf(),
+          new InetSocketAddress("localhost", 0), 100), getSelf());
+    }
+
+    @Override
+    public void onReceive(Object msg) throws Exception {
+      if (msg instanceof Bound) {
+        manager.tell(msg, getSelf());
+
+      } else if (msg instanceof CommandFailed) {
+        getContext().stop(getSelf());
+      
+      } else if (msg instanceof Connected) {
+        final Connected conn = (Connected) msg;
+        manager.tell(conn, getSelf());
+        final ActorRef handler = getContext().actorOf(
+            Props.create(SimplisticHandler.class));
+        getSender().tell(TcpMessage.register(handler), getSelf());
+      }
+    }
+    
+  }
+  //#server
+
+  static
+  //#simplistic-handler
+  public class SimplisticHandler extends UntypedActor {
+    @Override
+    public void onReceive(Object msg) throws Exception {
+      if (msg instanceof Received) {
+        final ByteString data = ((Received) msg).data();
+        System.out.println(data);
+        getSender().tell(TcpMessage.write(data), getSelf());
+      } else if (msg instanceof ConnectionClosed) {
+        getContext().stop(getSelf());
+      }
+    }
+  }
+  //#simplistic-handler
+  
+  static
+  //#client
+  public class Client extends UntypedActor {
+    
+    final InetSocketAddress remote;
+    final ActorRef listener;
+
+    public Client(InetSocketAddress remote, ActorRef listener) {
+      this.remote = remote;
+      this.listener = listener;
+      
+      final ActorRef tcp = Tcp.get(getContext().system()).manager();
+      tcp.tell(TcpMessage.connect(remote), getSelf());
+    }
+
+    @Override
+    public void onReceive(Object msg) throws Exception {
+      if (msg instanceof CommandFailed) {
+        listener.tell("failed", getSelf());
+        getContext().stop(getSelf());
+        
+      } else if (msg instanceof Connected) {
+        listener.tell(msg, getSelf());
+        getSender().tell(TcpMessage.register(getSelf()), getSelf());
+        getContext().become(connected(getSender()));
+      }
+    }
+
+    private Procedure<Object> connected(final ActorRef connection) {
+      return new Procedure<Object>() {
+        @Override
+        public void apply(Object msg) throws Exception {
+          
+          if (msg instanceof ByteString) {
+            connection.tell(TcpMessage.write((ByteString) msg), getSelf());
+        
+          } else if (msg instanceof CommandFailed) {
+            // OS kernel socket buffer was full
+          
+          } else if (msg instanceof Received) {
+            listener.tell(((Received) msg).data(), getSelf());
+          
+          } else if (msg.equals("close")) {
+            connection.tell(TcpMessage.close(), getSelf());
+          
+          } else if (msg instanceof ConnectionClosed) {
+            getContext().stop(getSelf());
+          }
+        }
+      };
+    }
+    
+  }
+  //#client
+  
+  private static ActorSystem system;
+  
+  @BeforeClass
+  public static void setup() {
+    system = ActorSystem.create("IODocTest", AkkaSpec.testConf());
+  }
+  
+  @AfterClass
+  public static void teardown() {
+    system.shutdown();
+  }
+  
+  @Test
+  public void testConnection() {
+    new JavaTestKit(system) {
+      {
+        @SuppressWarnings("unused")
+        final ActorRef server = system.actorOf(Props.create(Server.class, getRef()), "server1");
+        final InetSocketAddress listen = expectMsgClass(Bound.class).localAddress();
+        final ActorRef client = system.actorOf(Props.create(Client.class, listen, getRef()), "client1");
+        
+        final Connected c1 = expectMsgClass(Connected.class);
+        final Connected c2 = expectMsgClass(Connected.class);
+        assert c1.localAddress().equals(c2.remoteAddress());
+        assert c2.localAddress().equals(c1.remoteAddress());
+        
+        client.tell(ByteString.fromString("hello"), getRef());
+        final ByteString reply = expectMsgClass(ByteString.class);
+        assert reply.utf8String().equals("hello");
+        
+        watch(client);
+        client.tell("close", getRef());
+        expectTerminated(client);
+      }
+    };
+  }
+
+}

akka-docs/rst/java/code/docs/io/japi/SimpleEchoHandler.java

@@ -0,0 +1,130 @@
+/**
+ * Copyright (C) 2009-2013 Typesafe Inc. <http://www.typesafe.com>
+ */
+
+package docs.io.japi;
+
+import java.net.InetSocketAddress;
+import java.util.LinkedList;
+import java.util.Queue;
+
+import akka.actor.ActorRef;
+import akka.actor.UntypedActor;
+import akka.event.Logging;
+import akka.event.LoggingAdapter;
+import akka.io.Tcp.ConnectionClosed;
+import akka.io.Tcp.Received;
+import akka.io.TcpMessage;
+import akka.japi.Procedure;
+import akka.util.ByteString;
+
+//#simple-echo-handler
+public class SimpleEchoHandler extends UntypedActor {
+  
+  final LoggingAdapter log = Logging
+      .getLogger(getContext().system(), getSelf());
+
+  final ActorRef connection;
+  final InetSocketAddress remote;
+
+  public static final long maxStored = 100000000;
+  public static final long highWatermark = maxStored * 5 / 10;
+  public static final long lowWatermark = maxStored * 2 / 10;
+
+  public SimpleEchoHandler(ActorRef connection, InetSocketAddress remote) {
+    this.connection = connection;
+    this.remote = remote;
+
+    // sign death pact: this actor stops when the connection is closed
+    getContext().watch(connection);
+  }
+
+  @Override
+  public void onReceive(Object msg) throws Exception {
+    if (msg instanceof Received) {
+      final ByteString data = ((Received) msg).data();
+      buffer(data);
+      connection.tell(TcpMessage.write(data, ACK), getSelf());
+      // now switch behavior to “waiting for acknowledgement”
+      getContext().become(buffering, false);
+
+    } else if (msg instanceof ConnectionClosed) {
+      getContext().stop(getSelf());
+    }
+  }
+
+  private final Procedure<Object> buffering = new Procedure<Object>() {
+    @Override
+    public void apply(Object msg) throws Exception {
+      if (msg instanceof Received) {
+        buffer(((Received) msg).data());
+        
+      } else if (msg == ACK) {
+        acknowledge();
+
+      } else if (msg instanceof ConnectionClosed) {
+        if (((ConnectionClosed) msg).isPeerClosed()) {
+          closing = true;
+        } else {
+          // could also be ErrorClosed, in which case we just give up
+          getContext().stop(getSelf());
+        }
+      }
+    }
+  };
+
+  //#storage-omitted
+  public void postStop() {
+    log.info("transferred {} bytes from/to [{}]", transferred, remote);
+  }
+
+  private long transferred;
+  private long stored = 0;
+  private Queue<ByteString> storage = new LinkedList<ByteString>();
+
+  private boolean suspended = false;
+  private boolean closing = false;
+  
+  private final Object ACK = new Object();
+
+  //#simple-helpers
+  protected void buffer(ByteString data) {
+    storage.add(data);
+    stored += data.size();
+
+    if (stored > maxStored) {
+      log.warning("drop connection to [{}] (buffer overrun)", remote);
+      getContext().stop(getSelf());
+
+    } else if (stored > highWatermark) {
+      log.debug("suspending reading");
+      connection.tell(TcpMessage.suspendReading(), getSelf());
+      suspended = true;
+    }
+  }
+
+  protected void acknowledge() {
+    final ByteString acked = storage.remove();
+    stored -= acked.size();
+    transferred += acked.size();
+
+    if (suspended && stored < lowWatermark) {
+      log.debug("resuming reading");
+      connection.tell(TcpMessage.resumeReading(), getSelf());
+      suspended = false;
+    }
+    
+    if (storage.isEmpty()) {
+      if (closing) {
+        getContext().stop(getSelf());
+      } else {
+        getContext().unbecome();
+      }
+    } else {
+      connection.tell(TcpMessage.write(storage.peek(), ACK), getSelf());
+    }
+  }
+  //#simple-helpers
+  //#storage-omitted
+}
+//#simple-echo-handler

akka-docs/rst/java/code/docs/io/japi/Watcher.java

@@ -0,0 +1,34 @@
+package docs.io.japi;
+
+import java.util.concurrent.CountDownLatch;
+
+import akka.actor.ActorRef;
+import akka.actor.Terminated;
+import akka.actor.UntypedActor;
+
+public class Watcher extends UntypedActor {
+  
+  static public class Watch {
+    final ActorRef target;
+    public Watch(ActorRef target) {
+      this.target = target;
+    }
+  }
+  
+  final CountDownLatch latch;
+
+  public Watcher(CountDownLatch latch) {
+    this.latch = latch;
+  }
+
+  @Override
+  public void onReceive(Object msg) throws Exception {
+    if (msg instanceof Watch) {
+      getContext().watch(((Watch) msg).target);
+    } else if (msg instanceof Terminated) {
+      latch.countDown();
+      if (latch.getCount() == 0) getContext().stop(getSelf());
+    }
+  }
+
+}