[akka-user] Re: Memory Bounding Akka Streams

[Dagny T]

THANK YOU for your responses; and very helpful Endre to explain!

Need to Google up those JS interfaces for async and sync!

So, then, if you've got a system of remote services talking to each other;
you pretty much need to be coding to interfaces which support the Reactive 
protocol on (both) sides 
of the Network connection; otherwise the backpressure mechanism will fail, 
I guess.

Thanks!
D

On Wednesday, September 21, 2016 at 11:01:12 AM UTC-7, Dagny T wrote:
>
>
> Just wanted to check with folks if I had the correct implementation for 
> how to protect from blowing up memory when working with Akka Streams.
>
> I've merged a Lightbend Blog post's code, with the latest API changes for 
> Akka v2.4.9, and the latest documentation about buffered streams in the 
> v2.4.9 API Docs.
>
> However, none of those explain these questions I have.  Please see 
> question comments, regarding the code snippet below it!  THANKS in advance 
> for any insights!
>
> // TODO 3:  MODIFIED to calling buffered API within Graph mapping -- check 
> assumptions!
> //  - where INTERNAL Akka implementation calls onNext() to get next 
> BUFFERED batch,
> //so you don't have to worry about it as a DEV?
> //  - NUMERIC bound of 10 refers to NUMBER of elements (of possibly 
> complex types) on a
> //UNIFORM-ELEMENT-TYPED stream, rather than Bytes, right?
> //  - if source produces N < BUFFER_MAX elements; then those are 
> simply passed through the pipeline without
> //waiting to accumulate BUFFER_MAX elements
> //
>
>
> inputSource.buffer(10, OverflowStrategy.dropHead) ~> f1 ~> ...
>
>

-- 
>>  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.

[akka-user] Re: Best practices on message design

[Rob Crawford]

Look at Kamon for tracing:

http://kamon.io/core/tracing/core-concepts/


On Saturday, October 8, 2016 at 2:16:56 AM UTC-4, Shivakumar Ramagopal 
wrote:
>
> Hi,
>
> I'm looking for a guide for designing messages in an Akka system. My main 
> goal is to be able to trace messages as they pass from one actor to another 
> so as to be able to debug. Are there any best practices that you employ 
> when it comes to designing messages?
>
> Thanks,
> Shiva
>

-- 
>>  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.

[akka-user] Closing an outgoing TCP connection (akka.stream.scaladsl.Tcp().outgoingConnection) without terminating the ActorSystem

[Attila Szarvas]

I have a TCP client by connecting the Tcp().outgoingConnection to 
a Flow.fromSinkAndSourceMat where the sink and source are Sink.actorRef and 
Source.actorRef.

In general this works well. But I can't find a way to close the TCP 
connection from the client side. I can complete the stream by sending 
either a Success to the source ActorRef or a PoisonPill to both actors, but 
the TCP connection is left alive. I am even using the halfClose = false 
option. Watching the network communication no FIN is sent to the server and 
the server still believes the client to be connected.

Is this normal behavior? If I terminate the underlying ActorSystem the 
connection is closed, but I don't want to do that, as multiple clients are 
using the same system.

-- 
>>  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.

Re: [akka-user] How to handle web socket message to upload base64 encoded string of 10mb file

[Narayan Kumar]

thanks  for quick reply rafal .
Yes you hav mentioned right , I am getting the 'message does no arrive'  , 
but not any exception .

I have tried with  the TextMessage.Streamed , but no progress from here 
also . Please find the code below : -

case TextMessage.Streamed(stream) => {
  stream
.limit(1) // Max frames we are willing to wait for
.completionTimeout(50 seconds) // Max time until last frame
.runFold("")(_ + _) // Merges the frames
.flatMap { msg =>
  logger.info("Getting streamed message " + msg)
  val action = (parse(msg) \ (ACTION)).extractOpt[String]
  if (action.isDefined) {
action.get match {
  case UPDATE_USER_PROFILE =>
val userProfile = parse(msg).extractOpt[UserProfileJson].get
val cover = userProfile.cover
val picture = userProfile.picture
val uploadCoverFile = 
FileUploadUtility.decodeBase64String(cover.get)
val uploadPictureFile = 
FileUploadUtility.decodeBase64String(picture.get)
println(" 1 " + uploadCoverFile + " @@@  2 " + 
uploadPictureFile)
Future(TextMessage("File Uploaded"))
  case _ => Future(TextMessage(INVALID_ACTION))
}
  } else {
Future(TextMessage(INVALID_ACTION))
  }
}
}



Also I cannot test the case  properly cause the web socket client gets 
hang's every time I send the large file Json.,
and using the above code also we are not getting any response back . 
Please can you suggest any possible way to handle the scenarios.


Thanks in advance , waiting for positive early reply :) .



On Monday, October 10, 2016 at 3:12:36 PM UTC+5:30, Rafał Krzewski wrote:
>
> Please elaborate on "unable to handle it" -- are you getting an exception, 
> message does no arrive, something other?
> Also it would be helpful if you showed your code for TextMessage.Streamed 
> case because that's how large messages would show up. 
> I don't know the specifics but there appears to be a buffer in Akka 
> WebSockets client code: if the incoming message fits into this buffer in 
> full it's sent to the client code as Strict message, but when the message 
> is too large to fit in the buffer Akka switches to streaming mode: Streamed 
> message is sent to client code, carrying a stream that will deliver the 
> message contents in buffer-sized chunks.
>
> Cheers,
> aRafał
>
>
> W dniu poniedziałek, 10 października 2016 08:21:14 UTC+2 użytkownik 
> Narayan Kumar napisał:
>>
>>
>>
>> On Friday, October 7, 2016 at 7:52:03 PM UTC+5:30, √ wrote:
>>>
>>> Why are you assuming that it is a Strict message?
>>>
>>> On Fri, Oct 7, 2016 at 2:11 PM, Narayan Kumar  
>>> wrote:
>>>
 Hi everyone,
 Actually i was trying to handle a Web Socket message for base64 encoded 
 string of 10mb file.but unable to handle it.
 is there any way to handle large message please suggest ?

 Here is the code:

 def mediaUploadHandler: Flow[Message, Message, _] = {
 val (accountSource, accountQueue) = sourceQueue
 Flow[Message]
   .collect {
 case TextMessage.Strict(txt) ⇒ {
   logger.info(s"${phoneNumber}: Got the request. Now 
 redirecting to account api ")
   val userProfile = parse(txt).extractOpt[UserProfileJson]
   println("user profile is ", userProfile)
 }
 
 case _ => TextMessage(INVALID_ACTION)
   }
   .via(connectSource(accountSource)) // ... and route them through 
 the receiveNotification ...
   .map {
   case msg: String ⇒ {
 info(s"Huhh !! Why I am getting this message ${msg}")
 TextMessage.Strict(msg)
   }
 }
   }


 Thanks in advance!

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

>>>
>>>
>>>
>>> -- 
>>> Cheers,
>>> √
>>>
>>
>>
>>
>> Actually i have applied both approach of akka-http websocket 
>> "TextMessage.Strict(txt)" and "TextMessage.Streamed(stream)",but both 
>> approach didn't work.
>> is there is another approach to handle it please suggest ?
>>
>>
>>
>>
>>
>>
>>
>>  
>>
>

-- 
>>  Read the docs: http://akka.io/docs/
>>  Check the FAQ: 
>> 

Re: [akka-user] How to handle web socket message to upload base64 encoded string of 10mb file

[Rafał Krzewski]

Please elaborate on "unable to handle it" -- are you getting an exception, 
message does no arrive, something other?
Also it would be helpful if you showed your code for TextMessage.Streamed 
case because that's how large messages would show up. 
I don't know the specifics but there appears to be a buffer in Akka 
WebSockets client code: if the incoming message fits into this buffer in 
full it's sent to the client code as Strict message, but when the message 
is too large to fit in the buffer Akka switches to streaming mode: Streamed 
message is sent to client code, carrying a stream that will deliver the 
message contents in buffer-sized chunks.

Cheers,
aRafał


W dniu poniedziałek, 10 października 2016 08:21:14 UTC+2 użytkownik Narayan 
Kumar napisał:
>
>
>
> On Friday, October 7, 2016 at 7:52:03 PM UTC+5:30, √ wrote:
>>
>> Why are you assuming that it is a Strict message?
>>
>> On Fri, Oct 7, 2016 at 2:11 PM, Narayan Kumar  wrote:
>>
>>> Hi everyone,
>>> Actually i was trying to handle a Web Socket message for base64 encoded 
>>> string of 10mb file.but unable to handle it.
>>> is there any way to handle large message please suggest ?
>>>
>>> Here is the code:
>>>
>>> def mediaUploadHandler: Flow[Message, Message, _] = {
>>> val (accountSource, accountQueue) = sourceQueue
>>> Flow[Message]
>>>   .collect {
>>> case TextMessage.Strict(txt) ⇒ {
>>>   logger.info(s"${phoneNumber}: Got the request. Now 
>>> redirecting to account api ")
>>>   val userProfile = parse(txt).extractOpt[UserProfileJson]
>>>   println("user profile is ", userProfile)
>>> }
>>> 
>>> case _ => TextMessage(INVALID_ACTION)
>>>   }
>>>   .via(connectSource(accountSource)) // ... and route them through 
>>> the receiveNotification ...
>>>   .map {
>>>   case msg: String ⇒ {
>>> info(s"Huhh !! Why I am getting this message ${msg}")
>>> TextMessage.Strict(msg)
>>>   }
>>> }
>>>   }
>>>
>>>
>>> Thanks in advance!
>>>
>>> -- 
>>> >> 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+...@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.
>>>
>>
>>
>>
>> -- 
>> Cheers,
>> √
>>
>
>
>
> Actually i have applied both approach of akka-http websocket 
> "TextMessage.Strict(txt)" and "TextMessage.Streamed(stream)",but both 
> approach didn't work.
> is there is another approach to handle it please suggest ?
>
>
>
>
>
>
>
>  
>

-- 
>>  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.

[akka-user] Re: ('=') sign still not allowed in query string even with Uri.ParsingMode.Relaxed? (akka/akka#18479)

[André]

Hi Richard,

"GROUP=10380?page=2" isn't a well formed query and therefore can't be 
parsed even in relaxed mode. akka/akka#18479 
 provides a way to prevent parsing 
of the query component and to just look at the raw query string. You can 
access your query via the queryString() method and parse it youself.

HTH
André

On Sunday, October 9, 2016 at 6:38:34 PM UTC+2, Richard Imaoka wrote:
>
> Hi,
>
> While I tried to work on akka/akka-http#276 
> , and looked at 
> akka/akka#18479  (PR to fix 
> #18479 = akka/akka#18715 ), 
> I found that the following code:
>
> Query("GROUP=10380?page=2",mode = Uri.ParsingMode.Relaxed)
>
>
> throws an exception:
>
>   IllegalUriException: Illegal query: Invalid input '=', expected '+', 
> query-char, 'EOI', '&' or pct-encoded (line 1, column 17): GROUP=10380?page=
>
>
> Wasn't akka/akka#18479  (PR=
> akka/akka#18715 ) intended to 
> make this a valid query string, with Uri.ParsingMode.Relaxed ?
>
>
> *Additional Info*
> 1. I looked for a test case for this.
> According to UriSpec.scala 
> ,
>  
> Uri("?a=b=c").query() is still invalid, although query() method has default 
> mode = Uri.ParsingMode.Relaxed.
>
>
> 2. I didn't fully understand how parsing works  but probably it's 
> because CharacterClases.scala 
> 
> defines:
>
>   val `relaxed-query-char` = VCHAR -- "%&=#"
>
> where ('=') is made invalid?
>
> UriParser has 
>
> private[this] val `query-char` = uriParsingMode match {
>   case Uri.ParsingMode.Strict  ⇒ `strict-query-char`
>   case Uri.ParsingMode.Relaxed ⇒ `relaxed-query-char`
> }
>
> where `relaxed-query-char` is defined as above.
>
> Thanks,
> Richard
>
>

-- 
>>  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.

[akka-user] Re: akka.http.scaladsl.model.Uri's apply() methods should still take charset and mode parameters?

[André]

Hi Richard!

See [1] for the discussion related to your question.

[1] https://github.com/akka/akka/pull/18715#discussion-diff-41831981

Cheers
André

On Sunday, October 9, 2016 at 6:49:23 PM UTC+2, Richard Imaoka wrote:
>
> Hi,
>
> While I was trying to work on akka/akka-http#276 
> , I noticed something 
> strange.
>
> After akka/akka#18479  was fixed 
> (PR to fix #18479 = akka/akka#18715 
> ), now the following is possible.
> Isn't it non-intuitive that you can specify Uri.ParsingMode twice - once 
> in Uri() and secondly in query() method?
>
>
> val uri = Uri("http://localhost?a^=b;, mode=Uri.ParsingMode.Relaxed)
>
> //prints Some(b)
> println(uri.query(mode=Uri.ParsingMode.Relaxed).get("a^"))
>
> //IllegalUriException: Illegal query: Invalid input '^',
> // expected '+', '=', query-char, 'EOI', '&' or pct-encoded
> // (line 1, column 2): a^=b
> println(uri.query(mode=Uri.ParsingMode.Strict).get("a^"))
>
>
>
>
> If parsing a query string should be deferred as in akka/akka#18479 
> , should we not pass mode as well as 
> charset into akka.http.scaladsl.model.Uri companion object's apply() methods?
>
> Thanks,
> Richard
>
>
>
>

-- 
>>  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.

Re: [akka-user] How to handle web socket message to upload base64 encoded string of 10mb file

[Narayan Kumar]

On Friday, October 7, 2016 at 7:52:03 PM UTC+5:30, √ wrote:
>
> Why are you assuming that it is a Strict message?
>
> On Fri, Oct 7, 2016 at 2:11 PM, Narayan Kumar  > wrote:
>
>> Hi everyone,
>> Actually i was trying to handle a Web Socket message for base64 encoded 
>> string of 10mb file.but unable to handle it.
>> is there any way to handle large message please suggest ?
>>
>> Here is the code:
>>
>> def mediaUploadHandler: Flow[Message, Message, _] = {
>> val (accountSource, accountQueue) = sourceQueue
>> Flow[Message]
>>   .collect {
>> case TextMessage.Strict(txt) ⇒ {
>>   logger.info(s"${phoneNumber}: Got the request. Now 
>> redirecting to account api ")
>>   val userProfile = parse(txt).extractOpt[UserProfileJson]
>>   println("user profile is ", userProfile)
>> }
>> 
>> case _ => TextMessage(INVALID_ACTION)
>>   }
>>   .via(connectSource(accountSource)) // ... and route them through 
>> the receiveNotification ...
>>   .map {
>>   case msg: String ⇒ {
>> info(s"Huhh !! Why I am getting this message ${msg}")
>> TextMessage.Strict(msg)
>>   }
>> }
>>   }
>>
>>
>> Thanks in advance!
>>
>> -- 
>> >> 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+...@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.
>>
>
>
>
> -- 
> Cheers,
> √
>



Actually i have applied both approach of akka-http websocket 
"TextMessage.Strict(txt)" and "TextMessage.Streamed(stream)",but both 
approach didn't work.
is there is another approach to handle it please suggest ?







 

-- 
>>  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.