[akka-user] Alpakka - Contribution ideas
Hi, I've always known that the Akka community is very welcoming to the contributors. And Enno proved that to me again. I just made a small contribution towards the self-contained example (789 <https://github.com/akka/alpakka/pull/789>) in Alpakka and I am looking for ideas where I could learn and contribute more. TBH, I am just familiar with plain old Actors and everything else is a learning opportunity. Here's the list that I have in mind. Appreciate if you could let me know if these are not suitable for Alpakka contributions. If you have any other ideas else that involves reading and contributing to the Akka code base, please help. *Needs some homework but I can take a stab * Doobie connector (I believe it must be similar to Slick) HDFS connector (Already taken - Connector for HDFS #557) *Self-contained examples (Definitely possible)* Consume from RDBMS and route it to Solr (just like ES) Consume from Kafka and route it to RDBMS - Slick sink Consume from Kafka and push it to HBase Consume from Kafka and push it to Cassandra Thanks a ton, all. Cheers, Arun -- >>>>>>>>>> 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] Akka Kafka Stream - only once delivery
Thanks Michał.
On Wednesday, June 28, 2017 at 4:15:30 PM UTC+5:30, Michal Borowiecki wrote:
>
> If you need exactly once semantics against your target database, the
> common pattern is to store your last processed offset in that database
> transactionally together with your output records, instead of committing
> back to kafka. On startup you'd read the last offset from your database and
> seek to that offset.
>
> See section on external-offset-storage:
> http://doc.akka.io/docs/akka-stream-kafka/current/consumer.html#external-offset-storage
>
> You'd need to use the plainSource as the committableSource is
> fundamentally for at-least once delivery:
>
>
> http://doc.akka.io/docs/akka-stream-kafka/current/consumer.html#offset-storage-in-kafka
>
> Hope that helps,
>
> Michał
>
> On 28/06/17 11:20, Arun wrote:
>
> Hi,
>
> I am using Akka Kafka Consumer.committablePartitionedSource to stream
> messages from kafka and group them based on group key with groupedWithin .
> Grouped records should be sink into database and then it should able to
> commit offset.
>
> The code skeleton is as following:
>
> val consumerSettings = ConsumerSettings(system, new ByteArrayDeserializer,
> new ByteArrayDeserializer)
> .withBootstrapServers(kafkaServers)
> .withGroupId("testclientId")
> .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
> val kafkaSource:Source[CommittableMessage[Array[Byte], Array[Byte]], Control]=
>
>Consumer.committablePartitionedSource(consumerSettings,
> Subscriptions.topics(topicName)) .flatMapMerge(maxPartitions, _._2)
>
> val flow=Flow[In]
>
> .groupBy[K](maximumGroupSize, groupKeyFn)
> .map(e => groupKeyFn(e) -> map(e))
> .mergeSubstreams
> .groupedWithin(maximumGroupSize, FiniteDuration(10, TimeUnit.SECONDS))
>
> kafkaSource.via(flow).to(sinkToDBSaveOffSet).run
>
> The problem what we are facing, what if sinkToDB is successful, but
> SaveOffSet failed or what if system crashes before SaveOffSet is called, that
> means we will have duplicate records in the database. We don't want to do
> query operation before insert to find duplicate record.
>
> Is there any way we can use any fault tolerance strategy to avoid such
> duplicate message or saveOffset can become one transaction?
>
> Thanks & Regards,
>
> Arun
>
> --
> >>>>>>>>>> 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.
>
>
> --
> <http://www.openbet.com/> Michal Borowiecki
> Senior Software Engineer L4
> T: +44 208 742 1600
>
>
> +44 203 249 8448
>
>
>
> E: michal.b...@openbet.com
> W: www.openbet.com
> OpenBet Ltd
>
> Chiswick Park Building 9
>
> 566 Chiswick High Rd
>
> London
>
> W4 5XT
>
> UK
> <https://www.openbet.com/email_promo>
> This message is confidential and intended only for the addressee. If you
> have received this message in error, please immediately notify the
> ...@openbet.com and delete it from your system as well as
> any copies. The content of e-mails as well as traffic data may be monitored
> by OpenBet for employment and security purposes. To protect the environment
> please do not print this e-mail unless necessary. OpenBet Ltd. Registered
> Office: Chiswick Park Building 9, 566 Chiswick High Road, London, W4 5XT,
> United Kingdom. A company registered in England and Wales. Registered no.
> 3134634. VAT no. GB927523612
>
--
>>>>>>>>>> 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] Akka Kafka Stream - only once delivery
Hi,
I am using Akka Kafka Consumer.committablePartitionedSource to stream
messages from kafka and group them based on group key with groupedWithin .
Grouped records should be sink into database and then it should able to
commit offset.
The code skeleton is as following:
val consumerSettings = ConsumerSettings(system, new ByteArrayDeserializer, new
ByteArrayDeserializer)
.withBootstrapServers(kafkaServers)
.withGroupId("testclientId")
.withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
val kafkaSource:Source[CommittableMessage[Array[Byte], Array[Byte]], Control]=
Consumer.committablePartitionedSource(consumerSettings,
Subscriptions.topics(topicName)) .flatMapMerge(maxPartitions, _._2)
val flow=Flow[In]
.groupBy[K](maximumGroupSize, groupKeyFn)
.map(e => groupKeyFn(e) -> map(e))
.mergeSubstreams
.groupedWithin(maximumGroupSize, FiniteDuration(10, TimeUnit.SECONDS))
kafkaSource.via(flow).to(sinkToDBSaveOffSet).run
The problem what we are facing, what if sinkToDB is successful, but SaveOffSet
failed or what if system crashes before SaveOffSet is called, that means we
will have duplicate records in the database. We don't want to do query
operation before insert to find duplicate record.
Is there any way we can use any fault tolerance strategy to avoid such
duplicate message or saveOffset can become one transaction?
Thanks & Regards,
Arun
--
>>>>>>>>>> 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] No matching constructor found
class FileUploaderActor extends Actor{
override def receive: Receive = {
case UploadFile(billerId, filename, subCategory, count, dueDate) =>
val fileOberverActor = ActorSystem().actorOf(Props[FileObserverActor])
val billerData =
BillerFileUploadMetaData(billerId,filename,count,"ACTIVE",
new java.sql.Timestamp(new
java.util.Date().getTime),subCategory,dueDate)
val sparkSession = SparkContextHelper.sparkSession;
import sparkSession.implicits._
val rdd = sparkSession.sparkContext.parallelize(Seq(billerData))
val df = rdd.toDF()
df.write.format("org.apache.spark.sql.cassandra").options(Map("keyspace"
-> "billerplatform_schema", "table" ->
"biller_file_uploads")).mode(SaveMode.Append).save
fileOberverActor ! FileUploaded(filename, count)
}}
--
>> 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] Akka REST API - Path Param vs Headers
Hi,
I am using Akka HTTP routes to define Akka HTTP REST end points.
Are there any guidelines that says "we should not use path param, and try
to use header", if yes in what all cases it should be followed?
for example a URI like /students/{state}/{dept} vs using using URI
/staudents and pass state , dept as header value...
Thanks for your help.
Regards,
Arun
--
>>>>>>>>>> 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.kafka.KafkaConsumerActor$Internal$Stop$ Stop system
so far I could able be think create KafkaConsumerActor in Supervisor with supervisor strategy and use same with Consumer.committableExternalSource val consumer: ActorRef = context.actorOf(KafkaConsumerActor.props(consumerSettings)) Thanks Arun On Thursday, December 22, 2016 at 10:59:12 AM UTC+5:30, Arun wrote: > > Hi, > > I am using Akka Kafka stream ( > http://doc.akka.io/docs/akka-stream-kafka/current/home.html) to stream > messages from Kafka, if Kafka server is down, I starting getting following > error: > > Consumer interrupted with WakeupException after timeout. Message: null. > Current value of akka.kafka.consumer.wakeup-timeout is 3000 milliseconds > > Message [akka.kafka.KafkaConsumerActor$Internal$Stop$] from > Actor[akka://processor/deadLetters] to Actor[akka://processor/deadLetters] > to Actor[akka://processor/system/kafka-consumer-1#317821720] was not > delivered. [1] dead > > The current wakeup-timeout and max-wakeups are: > > wakeup-timeout = 3s > > # After exceeding maxinum wakeups the consumer will stop and the stage will > fail. > max-wakeups = 10 > > > How I can stop the actor system if it is not able to connect Kafka server > after 10 wakeups. > > > Thanks & Regards, > > Arun > > > > > > -- >>>>>>>>>> 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] akka.kafka.KafkaConsumerActor$Internal$Stop$ Stop system
Hi, I am using Akka Kafka stream (http://doc.akka.io/docs/akka-stream-kafka/current/home.html) to stream messages from Kafka, if Kafka server is down, I starting getting following error: Consumer interrupted with WakeupException after timeout. Message: null. Current value of akka.kafka.consumer.wakeup-timeout is 3000 milliseconds Message [akka.kafka.KafkaConsumerActor$Internal$Stop$] from Actor[akka://processor/deadLetters] to Actor[akka://processor/deadLetters] to Actor[akka://processor/system/kafka-consumer-1#317821720] was not delivered. [1] dead The current wakeup-timeout and max-wakeups are: wakeup-timeout = 3s # After exceeding maxinum wakeups the consumer will stop and the stage will fail. max-wakeups = 10 How I can stop the actor system if it is not able to connect Kafka server after 10 wakeups. Thanks & Regards, Arun -- >>>>>>>>>> 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] akka-kafka-stream "Offset Storage in Kafka" question/clarification
Thanks Patrik.
So am I safe to use mapAsync greater than 1 (>1) , still my offset will be
stored correctly?
what will happen if my stream fail/crash which offset will be processed..
Thanks,
Arun
On Saturday, August 27, 2016 at 12:26:30 PM UTC+5:30, Patrik Nordwall wrote:
>
> The commit commands are emitted to the underlying KafkaConsumer in right
> order. It is waiting for the results of the commits that are done in
> parallel, and will possibly increase total throughput. We observed better
> throughput in performance tests.
>
> It would be great to clarify this in docs. You are not the first one to
> ask. Would you like to open a pull request?
>
> /Patrik
> lör 27 aug. 2016 kl. 00:35 skrev Drew Kutcharian <dr...@venarc.com
> >:
>
>> Hi,
>>
>> In the "Offset Storage in Kafka” on
>> http://doc.akka.io/docs/akka-stream-kafka/current/consumer.html#offset-storage-in-kafka
>> there
>> is the following example:
>>
>> .groupedWithin(10, 5.seconds)
>> .map(group => group.foldLeft(CommittableOffsetBatch.empty) { (batch,
>> elem) => batch.updated(elem) })
>> .mapAsync(3)(_.commitScaladsl())
>>
>> Wouldn’t the ".mapAsync(3)(_.commitScaladsl())” cause the offsets to be
>> committed to Kafka out of order? Shouldn’t it be "mapAsync(1)”?
>>
>> Best,
>>
>> Drew
>>
>> --
>> >>>>>>>>>> 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.
>>
>
--
>>>>>>>>>> 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] Akka Http Logging with traceId
Thanks Yann and Jan.
The kamon looks clean way of implementation, the page also talks about
http://kamon.io/integrations/logback/mdc-in-an-asyncronous-environment/
the way you have describe in your blog as well.
I am going ahead with kamon.
Thanks to everyone.
Regards,
Arun
On Tuesday, September 13, 2016 at 2:02:46 PM UTC+5:30, Yann Simon wrote:
>
> I don't think so.
> The source thread is the thread used to handle the message, as the log is
> done in another actor (and maybe in another thread).
>
> Le mar. 13 sept. 2016 à 10:28, Arun <sethi...@gmail.com > a
> écrit :
>
>> Thanks Yann. Let me explore these options.
>>
>> As per akka documentation
>> http://doc.akka.io/docs/akka/2.4.10/scala/logging.html , can I use
>> "sourceThread"
>> as traceid?
>>
>> Regards,
>> Arun
>>
>> On Tuesday, September 13, 2016 at 1:23:48 PM UTC+5:30, Yann Simon wrote:
>>
>>> I wrote that some times ago:
>>> http://yanns.github.io/blog/2014/05/04/slf4j-mapped-diagnostic-context-mdc-with-play-framework/
>>> I use a somewhat modified version of that, but it works only with
>>> Futures.
>>> With actors (and akka-stream and akka-http), I think that you only have
>>> 2 options:
>>> - use kamon.io
>>> - propagating a "context" yourself, and using a custom Logger that reads
>>> from this context
>>>
>>> Yann
>>>
>>> Le mar. 13 sept. 2016 à 08:57, Arun <sethi...@gmail.com> a écrit :
>>>
>>>> Thanks Jan.
>>>>
>>>> I am looking for something similar to
>>>> http://code.hootsuite.com/logging-contextual-info-in-an-asynchronous-scala-application/
>>>>
>>>> , I could able to find that either by using custom akka dispatcher or
>>>> custom execution context, we can achieve the same, but I would prefer
>>>> anything I can do with minimal changes in akka default behaviour.
>>>>
>>>> Using MDC with Logback; the logback uses a ThreadLocal variable and
>>>> this works when one thread is used for one request, in case of Akka
>>>> asynchronous communication happens across multiple threads.
>>>>
>>>> Thanks
>>>> Arun
>>>>
>>>> On Tuesday, September 13, 2016 at 11:27:58 AM UTC+5:30, Jan Algermissen
>>>> wrote:
>>>>
>>>>> Hi Arun
>>>>>
>>>>> Take a look at kamon.io
>>>>>
>>>>> Jan
>>>>>
>>>>> Sent from my iPhone
>>>>>
>>>>
>>>>> On 13.09.2016, at 07:48, Arun <sethi...@gmail.com> wrote:
>>>>>
>>>>> Hi,
>>>>>
>>>>> We have a requirement where we need to log information with traceId
>>>>> (or unique identifier) for a given HTTP request across route and actors.
>>>>> This can help us to trace logging.
>>>>>
>>>>> The configuration is as following:
>>>>>
>>>>> akka {
>>>>> loggers = ["akka.event.slf4j.Slf4jLogger"]
>>>>> loglevel = "INFO"
>>>>> logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
>>>>>
>>>>> }
>>>>>
>>>>>
>>>>> and logback.xml is as following:
>>>>>
>>>>>
>>>>>
>>>>>
>>>>>
>>>>> true
>>>>>
>>>>> @timestamp
>>>>> msg
>>>>> [ignore]
>>>>> [ignore]
>>>>> logger
>>>>> [ignore]
>>>>>
>>>>> >>>> class="net.logstash.logback.stacktrace.ShortenedThrowableConverter">
>>>>> 80
>>>>> 2048
>>>>> 20
>>>>> true
>>>>>
>>>>>
>>>>>
>>>>>
>>>>>
>>>>>
>>>>>
>>>>>
>>>>>
>>>>> Please let me know how we can enable traceId from route to actor systems.
>>>>>
>>>>>
>>>>> Thanks
>>>>>
>>>>> Arun
>>
Re: [akka-user] Akka Http Logging with traceId
Thanks Yann. Let me explore these options.
As per akka
documentation http://doc.akka.io/docs/akka/2.4.10/scala/logging.html , can
I use "sourceThread" as traceid?
Regards,
Arun
On Tuesday, September 13, 2016 at 1:23:48 PM UTC+5:30, Yann Simon wrote:
>
> I wrote that some times ago:
> http://yanns.github.io/blog/2014/05/04/slf4j-mapped-diagnostic-context-mdc-with-play-framework/
> I use a somewhat modified version of that, but it works only with Futures.
> With actors (and akka-stream and akka-http), I think that you only have 2
> options:
> - use kamon.io
> - propagating a "context" yourself, and using a custom Logger that reads
> from this context
>
> Yann
>
> Le mar. 13 sept. 2016 à 08:57, Arun <sethi...@gmail.com > a
> écrit :
>
>> Thanks Jan.
>>
>> I am looking for something similar to
>> http://code.hootsuite.com/logging-contextual-info-in-an-asynchronous-scala-application/
>>
>> , I could able to find that either by using custom akka dispatcher or
>> custom execution context, we can achieve the same, but I would prefer
>> anything I can do with minimal changes in akka default behaviour.
>>
>> Using MDC with Logback; the logback uses a ThreadLocal variable and this
>> works when one thread is used for one request, in case of Akka asynchronous
>> communication happens across multiple threads.
>>
>> Thanks
>> Arun
>>
>> On Tuesday, September 13, 2016 at 11:27:58 AM UTC+5:30, Jan Algermissen
>> wrote:
>>
>>> Hi Arun
>>>
>>> Take a look at kamon.io
>>>
>>> Jan
>>>
>>> Sent from my iPhone
>>>
>>
>>> On 13.09.2016, at 07:48, Arun <sethi...@gmail.com> wrote:
>>>
>>> Hi,
>>>
>>> We have a requirement where we need to log information with traceId (or
>>> unique identifier) for a given HTTP request across route and actors. This
>>> can help us to trace logging.
>>>
>>> The configuration is as following:
>>>
>>> akka {
>>> loggers = ["akka.event.slf4j.Slf4jLogger"]
>>> loglevel = "INFO"
>>> logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
>>>
>>> }
>>>
>>>
>>> and logback.xml is as following:
>>>
>>>
>>>
>>>
>>>
>>> true
>>>
>>> @timestamp
>>> msg
>>> [ignore]
>>> [ignore]
>>> logger
>>> [ignore]
>>>
>>> >> class="net.logstash.logback.stacktrace.ShortenedThrowableConverter">
>>> 80
>>> 2048
>>> 20
>>> true
>>>
>>>
>>>
>>>
>>>
>>>
>>>
>>>
>>>
>>> Please let me know how we can enable traceId from route to actor systems.
>>>
>>>
>>> Thanks
>>>
>>> Arun
>>>
>>> --
>>> >>>>>>>>>> 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.
>>>
>>> --
>> >>>>>>>>>> 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.
>>
>
--
>>>>>>>>>> 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] Akka Http Logging with traceId
Thanks Jan.
I am looking for something similar
to
http://code.hootsuite.com/logging-contextual-info-in-an-asynchronous-scala-application/
, I could able to find that either by using custom akka dispatcher or
custom execution context, we can achieve the same, but I would prefer
anything I can do with minimal changes in akka default behaviour.
Using MDC with Logback; the logback uses a ThreadLocal variable and this
works when one thread is used for one request, in case of Akka asynchronous
communication happens across multiple threads.
Thanks
Arun
On Tuesday, September 13, 2016 at 11:27:58 AM UTC+5:30, Jan Algermissen
wrote:
>
> Hi Arun
>
> Take a look at kamon.io
>
> Jan
>
> Sent from my iPhone
>
> On 13.09.2016, at 07:48, Arun <sethi...@gmail.com > wrote:
>
> Hi,
>
> We have a requirement where we need to log information with traceId (or
> unique identifier) for a given HTTP request across route and actors. This
> can help us to trace logging.
>
> The configuration is as following:
>
> akka {
> loggers = ["akka.event.slf4j.Slf4jLogger"]
> loglevel = "INFO"
> logging-filter = "akka.event.slf4j.Slf4jLoggingFilter"
>
> }
>
>
> and logback.xml is as following:
>
>
>
>
>
> true
>
> @timestamp
> msg
> [ignore]
> [ignore]
> logger
> [ignore]
>
> class="net.logstash.logback.stacktrace.ShortenedThrowableConverter">
> 80
> 2048
> 20
> true
>
>
>
>
>
>
>
>
>
> Please let me know how we can enable traceId from route to actor systems.
>
>
> Thanks
>
> Arun
>
> --
> >>>>>>>>>> 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.
>
>
--
>>>>>>>>>> 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] swagger with akka http
Hi, We would like to document aka http services in swagger style. Any suggestion for the library. Thanks Arun -- >>>>>>>>>> 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: Intercept STDOUT of other process
Thanks. My question is more on STDOUT of any other process, specially if it is non java. On Saturday, March 19, 2016 at 6:54:51 AM UTC-5, Konrad Malawski wrote: > > Hi there, > > Please don't *immediately* cross post your questions here and stack > overflow > <http://stackoverflow.com/questions/36096694/intercept-stdout-and-stream-via-akka/36101376#36101376>, > > it makes it harder to track answered questions. It's ok to cross post if > after a while you did not receive an answer, but please don't do so > immediately. > > > In order to "intercept" stdout in Java you can setOut on the System object. > It takes a PrintWriter, which we are able to create by wrapping an > OutputStream "bridge" that Akka Streams provide, here's how: > > val is: OutputStream = StreamConverters.asOutputStream() >.to(Sink.foreach(println)) // your logic pipeline here >.run() > System.setOut(new PrintStream(is)) > > > -- > Konrad > Akka @ Lightbend > > W dniu sobota, 19 marca 2016 01:29:33 UTC+1 użytkownik Arun Sethia napisał: >> >> Hi, >> >> Is it possible to intercept STDOUT of any other process and stream them >> via akka streaming? >> >> Thanks >> Arun >> > -- >>>>>>>>>> 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] Intercept STDOUT of other process
Hi, Is it possible to intercept STDOUT of any other process and stream them via akka streaming? Thanks Arun -- >>>>>>>>>> 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] Monitoring Akka Streaming
Thanks. I would like to keep monitoring counters in the actorsystem, so those can be exposed via service to external world. On Thursday, March 10, 2016 at 5:48:12 PM UTC-6, rrodseth wrote: > > You can also use alsoTo to send stream elements to an actor or special > purpose Sink. > > On Thu, Mar 10, 2016 at 10:49 AM, Filippo De Luca <m...@filippodeluca.com > > wrote: > >> Hi, >> I suppose you can use map and call a external service for each message at >> defined stage. >> >> Even better you can build your own stage. >> >> On 10 March 2016 at 17:28, Arun Sethia <sethi...@gmail.com > >> wrote: >> >>> Hi, >>> >>> I have a requirement where we would like to know how many incoming >>> messages are processed by flow. We can use Materializer with AtomicLong >>> with Flow to do the same. >>> >>> Any other alternative or in-built functionality in akka-stream API? >>> >>> Thanks >>> Arun >>> >>> -- >>> >>>>>>>>>> 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. >>> >> >> >> >> -- >> >> >> [image: --] >> Filippo De Luca >> [image: http://]about.me/FilippoDeLuca >> <http://about.me/FilippoDeLuca?promo=email_sig> >> >> >> -- >> >>>>>>>>>> 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. >> > > -- >>>>>>>>>> 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] Re: Http Client with Source.tick
Super ... you guys are awesome ... more I am learning, it is becoming more
and more interesting.
I used recover, It worked well. Thanks a lot.
On Thursday, March 10, 2016 at 4:52:30 PM UTC-6, √ wrote:
>
>
> http://doc.akka.io/api/akka/2.4.2/?_ga=1.176202021.1085570125.1455222020#akka.stream.scaladsl.FlowOps
>
> --
> Cheers,
> √
> On Mar 10, 2016 11:43 PM, "Arun Sethia" <sethi...@gmail.com >
> wrote:
>
>> Thanks.
>>
>> please can you provide the link where I can see how to deal with such
>> errors.
>>
>> On Thursday, March 10, 2016 at 4:22:35 PM UTC-6, √ wrote:
>>>
>>> For each only deals with elements, not errors.
>>> Please consult the documentation for combinators that let you observe
>>> and manipulate errors.
>>>
>>> --
>>> Cheers,
>>> √
>>> On Mar 10, 2016 11:04 PM, "Arun Sethia" <sethi...@gmail.com> wrote:
>>>
>>>> httpSourceGraph1 is httpSourceGraph , but still same issue
>>>>
>>>>
>>>> val response= httpSourceGraph.via(httpConnFlow()).runForeach(println)
>>>>
>>>>
>>>> we can test same with any external internet URL also, the result remain
>>>> same.
>>>>
>>>>
>>>> On Thursday, March 10, 2016 at 3:55:22 PM UTC-6, Arun Sethia wrote:
>>>>>
>>>>> Hi,
>>>>>
>>>>> I am trying to connect http client to a http service exposed by
>>>>> server, the source should send request every 1 second for that I have
>>>>> crated following partial graphs:
>>>>>
>>>>>
>>>>> def httpSourceGraph() = {
>>>>> Source.fromGraph(GraphDSL.create() { implicit builder =>
>>>>> val sourceOutLet = builder.add(Source.tick(FiniteDuration(0,
>>>>> TimeUnit.SECONDS), FiniteDuration(1,
>>>>> TimeUnit.SECONDS),
>>>>> HttpRequest(uri ="/test", method = HttpMethods.GET))).out
>>>>> // expose outlet
>>>>> SourceShape(sourceOutLet)
>>>>> })
>>>>> }
>>>>>
>>>>>
>>>>> def httpConnFlow() = {
>>>>> Flow.fromGraph(GraphDSL.create() { implicit builder =>
>>>>>
>>>>> val httpSourceFlow = builder.add(Http(system).outgoingConnection(host
>>>>> = "localhost", port = 8080))
>>>>>
>>>>> FlowShape(httpSourceFlow.in, httpSourceFlow.out)
>>>>> })
>>>>> }
>>>>>
>>>>>
>>>>> the graph is composed as
>>>>>
>>>>>
>>>>> val response= httpSourceGraph1.via(httpConnFlow()).runForeach(println)
>>>>>
>>>>>
>>>>> if the http server (localhost:8080/test) is up and running, everything
>>>>> works fine, every 1 second I can see the response coming back from the
>>>>> server. I am not able to any response in case of either server is down
>>>>> or it goes down later.
>>>>>
>>>>>
>>>>> *I think it should give me following error:*
>>>>>
>>>>>
>>>>> akka.stream.StreamTcpException: Tcp command
>>>>> [Connect(localhost/127.0.0.1:8080,None,List(),Some(10 seconds),true)]
>>>>> failed
>>>>>
>>>>>
>>>>> Thanks for the help.
>>>>>
>>>>> -Arun
>>>>>
>>>>>
>>>>>
>>>>>
>>>>> --
>>>> >>>>>>>>>> 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.
>>>>
>>> --
>> >>>>>>>>>> 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.
>>
>
--
>>>>>>>>>> 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] Re: Http Client with Source.tick
Thanks.
please can you provide the link where I can see how to deal with such
errors.
On Thursday, March 10, 2016 at 4:22:35 PM UTC-6, √ wrote:
>
> For each only deals with elements, not errors.
> Please consult the documentation for combinators that let you observe and
> manipulate errors.
>
> --
> Cheers,
> √
> On Mar 10, 2016 11:04 PM, "Arun Sethia" <sethi...@gmail.com >
> wrote:
>
>> httpSourceGraph1 is httpSourceGraph , but still same issue
>>
>>
>> val response= httpSourceGraph.via(httpConnFlow()).runForeach(println)
>>
>>
>> we can test same with any external internet URL also, the result remain
>> same.
>>
>>
>> On Thursday, March 10, 2016 at 3:55:22 PM UTC-6, Arun Sethia wrote:
>>>
>>> Hi,
>>>
>>> I am trying to connect http client to a http service exposed by server,
>>> the source should send request every 1 second for that I have crated
>>> following partial graphs:
>>>
>>>
>>> def httpSourceGraph() = {
>>> Source.fromGraph(GraphDSL.create() { implicit builder =>
>>> val sourceOutLet = builder.add(Source.tick(FiniteDuration(0,
>>> TimeUnit.SECONDS), FiniteDuration(1,
>>> TimeUnit.SECONDS),
>>> HttpRequest(uri ="/test", method = HttpMethods.GET))).out
>>> // expose outlet
>>> SourceShape(sourceOutLet)
>>> })
>>> }
>>>
>>>
>>> def httpConnFlow() = {
>>> Flow.fromGraph(GraphDSL.create() { implicit builder =>
>>>
>>> val httpSourceFlow = builder.add(Http(system).outgoingConnection(host =
>>> "localhost", port = 8080))
>>>
>>> FlowShape(httpSourceFlow.in, httpSourceFlow.out)
>>> })
>>> }
>>>
>>>
>>> the graph is composed as
>>>
>>>
>>> val response= httpSourceGraph1.via(httpConnFlow()).runForeach(println)
>>>
>>>
>>> if the http server (localhost:8080/test) is up and running, everything
>>> works fine, every 1 second I can see the response coming back from the
>>> server. I am not able to any response in case of either server is down or
>>> it goes down later.
>>>
>>>
>>> *I think it should give me following error:*
>>>
>>>
>>> akka.stream.StreamTcpException: Tcp command
>>> [Connect(localhost/127.0.0.1:8080,None,List(),Some(10 seconds),true)] failed
>>>
>>>
>>> Thanks for the help.
>>>
>>> -Arun
>>>
>>>
>>>
>>>
>>> --
>> >>>>>>>>>> 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.
>>
>
--
>>>>>>>>>> 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] Http Client with Source.tick
Hi,
I am trying to connect http client to a http service exposed by server, the
source should send request every 1 second for that I have crated following
partial graphs:
def httpSourceGraph() = {
Source.fromGraph(GraphDSL.create() { implicit builder =>
val sourceOutLet = builder.add(Source.tick(FiniteDuration(0,
TimeUnit.SECONDS), FiniteDuration(1,
TimeUnit.SECONDS),
HttpRequest(uri ="/test", method = HttpMethods.GET))).out
// expose outlet
SourceShape(sourceOutLet)
})
}
def httpConnFlow() = {
Flow.fromGraph(GraphDSL.create() { implicit builder =>
val httpSourceFlow = builder.add(Http(system).outgoingConnection(host =
"localhost", port = 8080))
FlowShape(httpSourceFlow.in, httpSourceFlow.out)
})
}
the graph is composed as
val response= httpSourceGraph1.via(httpConnFlow()).runForeach(println)
if the http server (localhost:8080/test) is up and running, everything works
fine, every 1 second I can see the response coming back from the server. I am
not able to any response in case of either server is down or it goes down
later.
*I think it should give me following error:*
akka.stream.StreamTcpException: Tcp command
[Connect(localhost/127.0.0.1:8080,None,List(),Some(10 seconds),true)] failed
Thanks for the help.
-Arun
--
>>>>>>>>>> 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] Monitoring Akka Streaming
Hi, I have a requirement where we would like to know how many incoming messages are processed by flow. We can use Materializer with AtomicLong with Flow to do the same. Any other alternative or in-built functionality in akka-stream API? Thanks Arun -- >>>>>>>>>> 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] Source Repeat given interval
thanks a lot. Cool.. I should have looked into http://doc.akka.io/api/akka-stream-and-http-experimental/2.0/#akka.stream.scaladsl.Source val input:List[Int]=List(1,2,3) val source1 = Source.tick(FiniteDuration(1,TimeUnit.SECONDS),FiniteDuration(1,TimeUnit.SECONDS),input) source1.mapConcat[Int](_.toList).via(flow).runForeach(println) What you would suggest if List[HttpRequest] and would like to use Flow as Http(system).outgoingConnection(host = "abs", port = 8080) can I use same as above or any better approach. On Monday, March 7, 2016 at 3:29:43 PM UTC-6, √ wrote: > > mapConcat(identity) > > -- > Cheers, > √ > On Mar 7, 2016 10:26 PM, "Arun Sethia" <sethi...@gmail.com > > wrote: > >> Hi, >> >> I have requirement where source should repeat infinite times after given >> interval, for example: >> >> val input:List[Int]=List(1,2,3) >> >> >> val flow=Flow[Int].map(x=> x * 2) >> >> >> Source(input.toList).via(*flow)*.runForeach(println) >> >> >> The source "input" should repeat every 1 second. >> >> >> I tried to use val source = >> Source.tick(FiniteDuration(1,TimeUnit.SECONDS),FiniteDuration(1,TimeUnit.SECONDS),input), >> but this will provide out stream as List. >> >> >> Thanks >> >> Arun >> >> -- >> >>>>>>>>>> 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. >> > -- >>>>>>>>>> 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] Source Repeat given interval
Hi, I have requirement where source should repeat infinite times after given interval, for example: val input:List[Int]=List(1,2,3) val flow=Flow[Int].map(x=> x * 2) Source(input.toList).via(*flow)*.runForeach(println) The source "input" should repeat every 1 second. I tried to use val source = Source.tick(FiniteDuration(1,TimeUnit.SECONDS),FiniteDuration(1,TimeUnit.SECONDS),input), but this will provide out stream as List. Thanks Arun -- >>>>>>>>>> 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: how to define materialized value
Awesome thanks a lot.
On Sunday, March 6, 2016 at 7:20:43 AM UTC-6, Rafał Krzewski wrote:
>
> Arun,
>
> a little correction:
>
> val runnableGraph =
> source.viaMat(counter[Int])(Keep.right).toMat(sink)(Keep.both)
>
> And subsequently:
>
> val (counter, futureSum) = runnableGraph.run()
>
> Graph outlets are always streams. You need to connect them to a Sink
> (through intervening Flows or more complex Graphs, as necessary) in order
> to create a RunnableGraph. Materialized values are the other things used
> to connect the RunnableGraph to the outside world that are *not* streams.
>
> For example Sink.fold creates a stream element that is (obviously) a Sink.
> It does not have any stream outlets. However it provides a materialized
> value Future[U] that is completed when the Sink's inlet stream is
> exhausted. This is how a running stream can communicate it's successful
> completion or failure to the outside world.
>
> Another example is Source.actorPublisher: you provide it with Props for
> an Actor that implements ActorPublisher contract. When materializing the
> stream, the Source will instantiate the Actor and return it's ActorRef as
> a materialized value. The Actor is internal to the stream but you can use
> the ActorRef as an interface from the outside world into the stream: send
> messages (using your own protocol) to be passed to the Source's outlet,
> according to demand from downstream. The tricky part is that such gateway
> Actor must manage buffering and/or backpressure on it's own!
>
> Besides that, you can use materialized values to monitor stream execution
> from the outside, like in the Counter example above or
> https://github.com/akka/akka/pull/19836 or to interrupt a stream that
> would otherwise run for a long (or unlimited) time:
> https://github.com/rkrzewski/akka-cluster-etcd/blob/master/etcd-client/src/main/scala/pl/caltha/akka/streams/FlowBreaker.scala
>
> Cheers,
> Rafał
>
> W dniu niedziela, 6 marca 2016 08:43:10 UTC+1 użytkownik Arun Sethia
> napisał:
>>
>> Thanks Rafal.
>>
>> Based on this I tried to make sample code, where I would like to count
>> number of elements being processed and their sum:
>>
>> val source = Source (1 to 5).filter(x=> x%2==0)
>>
>> val sink:Sink[Int, Future[Int]]=Sink.fold[Int,Int](0)(_ + _)
>>
>> val runnableGraph = source.via(counter[Int]).toMat(sink)(Keep.both)
>>
>> val result=runnableGraph.run()
>>
>>
>> def counter[T]: Flow[T, T, Counter] = {
>> val internalCounter = new AtomicLong(0)
>> Flow[T].map{ elem ⇒
>> internalCounter.incrementAndGet()
>> elem
>> }.mapMaterializedValue(_ ⇒ new Counter{
>> override def get = internalCounter.get
>> })
>> }
>>
>>
>>
>> 1. using Keep.both, result should able to return me count and sum, but it is
>> not?
>>
>> 2. How materialize values are different than "out"? I am not able to
>> visualize the difference between materialize values and out?
>>
>> Thanks
>> Arun
>>
>>
>>
>> On Saturday, March 5, 2016 at 6:02:56 PM UTC-6, Arun Sethia wrote:
>>>
>>> Hi,
>>>
>>> can some explain what does it mean of materialized value ? I have see
>>> documentation at
>>> http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0.3/scala/stream-quickstart.html#transforming-and-consuming-simple-streams
>>>
>>>
>>> I am not sure how Flow can define materialize type, for example the
>>> following code has Input - Tweet, output - Int but Mat is Unit. I would
>>> like to see how someone can define Mat as Int or any example where Flow or
>>> source is defining Mat other than Unit.
>>>
>>> - val count: Flow[Tweet, Int, Unit] = Flow[Tweet].map(_ => 1)
>>>
>>>
>>>
>>> It is quite confusing for me to understand difference between "out" and
>>> "Mat".
>>>
>>>
>>> Thanks
>>>
>>> As
>>>
>>>
--
>>>>>>>>>> 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: how to define materialized value
the given same code: val source = Source (1 to 5).filter(x=> x%2==0) val flow:Flow[Int,Int,Unit]=Flow[Int].map(x=> x * 2) val sink:Sink[Int, Future[Int]]=Sink.fold[Int,Int](0)(_ + _) val runnableGraph = source.via(flow).toMat(sink)(Keep.both) I am not sure what is use of using Keep.both vs Keep.left, I thought If I use keep.both, will able to get values for flow and sink as tuple. On Saturday, March 5, 2016 at 6:02:56 PM UTC-6, Arun Sethia wrote: > > Hi, > > can some explain what does it mean of materialized value ? I have see > documentation at > http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0.3/scala/stream-quickstart.html#transforming-and-consuming-simple-streams > > > I am not sure how Flow can define materialize type, for example the > following code has Input - Tweet, output - Int but Mat is Unit. I would > like to see how someone can define Mat as Int or any example where Flow or > source is defining Mat other than Unit. > > - val count: Flow[Tweet, Int, Unit] = Flow[Tweet].map(_ => 1) > > > > It is quite confusing for me to understand difference between "out" and > "Mat". > > > Thanks > > As > > -- >>>>>>>>>> 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: how to define materialized value
more Source[+Out, +Mat],Flow[-In, +Out, +Mat] and Sink[-In, +Mat] , in all cases what is +Mat type and how I can define one such , if possible any example will be great. On Saturday, March 5, 2016 at 6:02:56 PM UTC-6, Arun Sethia wrote: > > Hi, > > can some explain what does it mean of materialized value ? I have see > documentation at > http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0.3/scala/stream-quickstart.html#transforming-and-consuming-simple-streams > > > I am not sure how Flow can define materialize type, for example the > following code has Input - Tweet, output - Int but Mat is Unit. I would > like to see how someone can define Mat as Int or any example where Flow or > source is defining Mat other than Unit. > > - val count: Flow[Tweet, Int, Unit] = Flow[Tweet].map(_ => 1) > > > > It is quite confusing for me to understand difference between "out" and > "Mat". > > > Thanks > > As > > -- >>>>>>>>>> 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: how to define materialized value
Thanks Rafal.
Based on this I tried to make sample code, where I would like to count
number of elements being processed and their sum:
val source = Source (1 to 5).filter(x=> x%2==0)
val sink:Sink[Int, Future[Int]]=Sink.fold[Int,Int](0)(_ + _)
val runnableGraph = source.via(counter[Int]).toMat(sink)(Keep.both)
val result=runnableGraph.run()
def counter[T]: Flow[T, T, Counter] = {
val internalCounter = new AtomicLong(0)
Flow[T].map{ elem ⇒
internalCounter.incrementAndGet()
elem
}.mapMaterializedValue(_ ⇒ new Counter{
override def get = internalCounter.get
})
}
1. using Keep.both, result should able to return me count and sum, but it is
not?
2. How materialize values are different than "out"? I am not able to visualize
the difference between materialize values and out?
Thanks
Arun
On Saturday, March 5, 2016 at 6:02:56 PM UTC-6, Arun Sethia wrote:
>
> Hi,
>
> can some explain what does it mean of materialized value ? I have see
> documentation at
> http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0.3/scala/stream-quickstart.html#transforming-and-consuming-simple-streams
>
>
> I am not sure how Flow can define materialize type, for example the
> following code has Input - Tweet, output - Int but Mat is Unit. I would
> like to see how someone can define Mat as Int or any example where Flow or
> source is defining Mat other than Unit.
>
> - val count: Flow[Tweet, Int, Unit] = Flow[Tweet].map(_ => 1)
>
>
>
> It is quite confusing for me to understand difference between "out" and
> "Mat".
>
>
> Thanks
>
> As
>
>
--
>>>>>>>>>> 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] how to define materialized value
Hi, can some explain what does it mean of materialized value ? I have see documentation at http://doc.akka.io/docs/akka-stream-and-http-experimental/2.0.3/scala/stream-quickstart.html#transforming-and-consuming-simple-streams I am not sure how Flow can define materialize type, for example the following code has Input - Tweet, output - Int but Mat is Unit. I would like to see how someone can define Mat as Int or any example where Flow or source is defining Mat other than Unit. - val count: Flow[Tweet, Int, Unit] = Flow[Tweet].map(_ => 1) It is quite confusing for me to understand difference between "out" and "Mat". Thanks As -- >> 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] Context.actorOf vs system.actorOf
I have been writing a few blogs on Actor to log my understanding of Akka. I happen to take a simple Student Teacher interaction as an analogy to explain how Actor message passing works. The students sends a mail (immutable) and the teacher receives the mail from his mailbox at his own time. Question 1 1. Considering the TeacherActor isn’t a subtask of the Student Actor, I personally think that using context.actorOf is not the right approach to creating the TeacherActor. Also, it doesn’t sound right semantically. 2. If the TeacherActor is supposed to be a top-level Actor, I end up creating one top-level Actor per request – which is wrong again as I understand that creating top-level actors are expensive. (I understand that this is because it creates a new ThreadPool per top-level Actor?). I am thinking of creating the TeacherActor upfront and pass the ActorRef to the constructor of the Student. Question 2 Also, as far as I understand, 1. If you plan to create a child actor, the general guideline is to pass the Props to the creating actor’s constructor – so that it resurrect the child Actor if needed. 2. If you plan to use a top-level Actor, pass in the actual ActorRef reference (mostly these are supervisors/routers). Is this a sane understanding? If anyone is curious, the url is http://rerun.me -- 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 http://groups.google.com/group/akka-user. For more options, visit https://groups.google.com/d/optout.
