Connect two TCP sockets

[Mike Limansky]

Hi all,

  I'm very new to Akka Streams, trying to get the ideas behind the streams and how to use them. I'd like to make a very simple thing: to connect two TCP server sockets, but I get stuck with that.

  If I have server and client connections, it more or less clear for me:

  val conn = Tcp().bind("localhost", 1234)

  val out = Tcp().outgoingConnection("localhost", 2345)
 

  val handler = Sink.foreach[Tcp.IncomingConnection] { c =>

    println("Got connection from " + c.remoteAddress)

    c.handleWith(out)

  }

  conn.to(handler).run()

  This working for one connection. Could anybody give me a clue how to get it work with more that one incoming connections (I suppose I need to use Merge, or Source.combine)?

  And the second question. As I understand this works in one direction (from server socket to client). How can I use BidiFlow in such situation?

  Finally, If I have two server connections I have no idea how to achieve same result. Should I use mutable state to store incoming connections?
 
--
BR,
Mike.

[Johan Andrén]

Hi Mike,

The problem is that you create one outgoing connection, but you cannot use that more than once. If you move that logic to create a new outgoing connection on every new incoming it will work better, so that would be inside of your foreach.

Please take a look at the docs for doing IO with Streams here: http://doc.akka.io/docs/akka/2.4.4/scala/stream/stream-io.html#Streaming_TCP 

(I think that will answer most of your questions)

--

Johan Andrén

Akka Team, Lightbend Inc.

[Mike Limansky]

Hi Johan,

  I meant I'd like to merge data from all incoming connections to one output.



вторник, 19 апреля 2016 г., 21:50:18 UTC+3 пользователь Johan Andrén написал:

[Johan Andrén]

Hi Mike,

There currently is no such stage, something like dynamic pub-sub (we have some ideas that you can read about in this ticket: https://github.com/akka/akka/issues/19478), so to do that you would have to either write your own stage for it or do it with actors (but then you loose backpressure).

--

Johan Andrén

Akka Team, Lightbend Inc.

 

--
>>>>>>>>>> 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 a topic in the Google Groups "Akka User List" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/akka-user/AvFbNTKMEdo/unsubscribe.
To unsubscribe from this group and all its topics, send an email to akka-user+...@googlegroups.com.
To post to this group, send email to akka...@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.

[Mike Limansky]

Hi Johan,

  Thanks a lot. Now this part is clear for me.

среда, 20 апреля 2016 г., 11:05:59 UTC+3 пользователь Johan Andrén написал:

[Endre Varga]

You can use Sink.queue to get a future based queue API that is backpressured. Then you can just combine with mapAsync on the sockets side to get backpressured merges. Not exactly the same as Hub would work, but already works for this simple case.

-Endre

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+...@googlegroups.com.