This question got an answer in
https://groups.google.com/forum/#!topic/akka-user/qpqWePkADwU
On Tuesday, August 25, 2015 at 2:40:11 AM UTC+2, Simon Schäfer wrote:
>
> I have a server, which on the one hand needs to send data and on the other
> hands needs to receive data. So far I was unable to implement this.
>
> The only code example, which I found, is this:
>
> val tcpFlow = Tcp().outgoingConnection(host, port)
> val bytes: IndexedSeq[ByteString] = ???
> val resp = Source(bytes).via(tcpFlow).runFold(ByteString.empty)(_++_)
> resp onComplete { ... }
>
> I don't understand how this works. Why can I receive a response from the
> other side, when I create an outgoing connection? Just because tcpFlow is a
> Flow? Why is it then called outgoingConnection and not just connection when
> it is bidirectional anyway?
>
> Also, this example doesn't help me because I not only need to send a
> request and receive a response after that, but I also need to receive
> messages without a prior request for it. How do I implement this?
>
> Basically, my interface is
>
> def send(request: Request): Future[Response]
> def register(eventType: Sting): Source[Event, Unit]
>
> where the latter should give me all events of the given type whenever one
> is created. But I just can't figure out how to get from a TCP connection to
> such a Source. I have this:
>
> implicit val m = ActorMaterializer()
> import system.dispatcher
>
> val flow = Flow[ByteString].fold(ByteString.empty)(_++_).map { resp ⇒
> // here I receive the data and need somehow need to convert it to a
> Source
> resp
> }
>
> val sink = Sink.foreach[Tcp.IncomingConnection] { connection ⇒
> connection.handleWith(flow)
> }
>
> val connections = Tcp().bind(host, port)
> val binding = connections.to(sink).run()
> binding onComplete {
> case Success(binding) ⇒
> println("binding created" + binding.localAddress)
> case Failure(f) ⇒
> f.printStackTrace()
> }
>
> The flow receives the data, but how could I put it into a Source in order
> to return it to the caller? With this setup I also can't send anymore:
>
> // inside of the `send` method
> val tcpFlow = Tcp().outgoingConnection(host, port)
> Source(bytes).via(tcpFlow).to(Sink.ignore).run()
>
> This way my `flow` receives everything that is send to `tcpFlow`.
>
> I'm completely confused about how to put things together. Can someone tell
> me how I can do what I want?
>
--
>>>>>>>>>> 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 http://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
