scalaz-stream consume stream based on computed value

Question

I've got two streams and I want to be able to consume only one based on a computation that I run every x seconds.

I think I basically need to create a third tick stream - something like every(3.seconds) - that does the computation and then come up with sort of a switch between the other two.

I'm kind of stuck here (and I've only just started fooling around with scalaz-stream).

Thanks!

Answer 1

There are several ways we can approach this problem. One way to approach it is using awakeEvery. For concrete example, see here.

To describe the example briefly, consider that we would like to query twitter in every 5 sec and get the tweets and perform sentiment analysis. We can compose this pipeline as follows:

val source = 
    awakeEvery(5 seconds) |> buildTwitterQuery(query) through queryChannel flatMap {
        Process emitAll _ 
    }

Note that the queryChannel can be stated as follows.

def statusTask(query: Query): Task[List[Status]] = Task {
      twitterClient.search(query).getTweets.toList
}

val queryChannel: Channel[Task, Query, List[Status]] = channel lift statusTask

Let me know if you have any question. As stated earlier, for the complete example, see this.

I hope it helps!