I have a (stashed) PR implementing exactly what you ask for here, however
we first need to ship the new JavaDSL before I can revive this PR.
My work in progress PR was here: https://github.com/akka/akka/pull/18745
People have actually copied it and used on prod I hear ;-)
Anyway, it's a feature we will provide in the Routing DSL, I'm just not
sure when exactly.
On Sun, Apr 3, 2016 at 5:41 PM, RANDOM MOD 2 <cesar4...@gmail.com> wrote:
> Hello,
>
>
>
> 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?
>
>
> 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();
>
> });
> }
>
> }
>
> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ:
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
--
Cheers,
Konrad 'ktoso' Malawski
Akka <http://akka.io/> @ Lightbend <http://lightbend.com/>
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
