CODEWITHSUNDEEP
javastreamakkaakka-stream
broga
broga

Reputation: 31

Akka stream of objects over http

I have got a piece of code (see bellow) which spawns a server that echoes every stream of ByteString it receives from port 6001. The example also defines a client that connects to the server and sends a stream of ByteString containing a list of characters from letter 'a' to 'z'.

My question at this point is, does akka offer a way to send and receive a Stream of objects instead of ByStreams over http? For instance, objects of class Client.

If so, how could I send and receive such a stream of objects? Could you provide me a snippet that shows how to carry it out?

Akka documentation is not user-friendly for non-toy examples...

Thanks for your help

public class TcpEcho {

/**
 * Use without parameters to start both client and server.
 *
 * Use parameters `server 0.0.0.0 6001` to start server listening on port
 * 6001.
 *
 * Use parameters `client 127.0.0.1 6001` to start client connecting to
 * server on 127.0.0.1:6001.
 *
 */
public static void main(String[] args) throws IOException {
    if (args.length == 0) {
        ActorSystem system = ActorSystem.create("ClientAndServer");
        InetSocketAddress serverAddress = new InetSocketAddress("127.0.0.1", 6000);
        server(system, serverAddress);
        client(system, serverAddress);
    } else {
        InetSocketAddress serverAddress;
        if (args.length == 3) {
            serverAddress = new InetSocketAddress(args[1], Integer.valueOf(args[2]));
        } else {
            serverAddress = new InetSocketAddress("127.0.0.1", 6000);
        }
        if (args[0].equals("server")) {
            ActorSystem system = ActorSystem.create("Server");
            server(system, serverAddress);
        } else if (args[0].equals("client")) {
            ActorSystem system = ActorSystem.create("Client");
            client(system, serverAddress);
        }
    }
}

public static void server(ActorSystem system, InetSocketAddress serverAddress) {
    final ActorMaterializer materializer = ActorMaterializer.create(system);

    final Sink<IncomingConnection, CompletionStage<Done>> handler = Sink.foreach(conn -> {
        System.out.println("Client connected from: " + conn.remoteAddress());
        conn.handleWith(Flow.<ByteString> create(), materializer);
    });

    final CompletionStage<ServerBinding> bindingFuture = Tcp.get(system)
            .bind(serverAddress.getHostString(), serverAddress.getPort()).to(handler).run(materializer);

    bindingFuture.whenComplete((binding, throwable) -> {
        System.out.println("Server started, listening on: " + binding.localAddress());
    });

    bindingFuture.exceptionally(e -> {
        System.err.println("Server could not bind to " + serverAddress + " : " + e.getMessage());
        system.terminate();
        return null;
    });

}

public static void client(ActorSystem system, InetSocketAddress serverAddress) {
    final ActorMaterializer materializer = ActorMaterializer.create(system);

    final List<ByteString> testInput = new ArrayList<>();
    for (char c = 'a'; c <= 'z'; c++) {
        testInput.add(ByteString.fromString(String.valueOf(c)));
    }

    Source<ByteString, NotUsed> responseStream = Source.from(testInput)
            .via(Tcp.get(system).outgoingConnection(serverAddress.getHostString(), serverAddress.getPort()));

    CompletionStage<ByteString> result = responseStream.runFold(ByteString.empty(), (acc, in) -> acc.concat(in),
            materializer);

    result.whenComplete((success, failure) -> {

        if (failure != null) {
            System.err.println("Failure: " + failure.getMessage());
        } else {
            System.out.println("Result: " + success.utf8String());
        }
        System.out.println("Shutting down client");
        system.terminate();

    });
}

}

Upvotes: 2

Views: 377

Answers (1)

Samuel Tardieu
Samuel Tardieu

Reputation: 2081

akka.stream.{javadsl,scaladsl}.Framing contains utilities to help you build consistent messages. For example, you can send your messages through Framing.simpleFramingProtocolEncoder(maxLength) to automatically add length information to them. On the other end, Framing.simpleFramingProtocolDecoder(maxLength) will take care of decoding the message according to its enclosed length information.

If you want to manipulate plain objects, you just have to serialize them into a ByteString before sending them through the encoder, and deserialize them from the ByteString after receiving their representation from the decoder.

Upvotes: 2

Related Questions