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 [email protected].
To post to this group, send email to [email protected].
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
