# Creating a channel from exhaustible source

**URL:** <https://discuss.kotlinlang.org/t/creating-a-channel-from-exhaustible-source/7464>\
**Category:** Support\
**Created:** [April 18, 2018, 6:04pm UTC](https://discuss.kotlinlang.org/t/creating-a-channel-from-exhaustible-source/7464 "2018-04-18T18:04:26Z")\
**Posts on this page:** 3\
**Page:** 1

<div class="post-metadata">

**Author:** ![oripwk](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/oripwk/32/5028_2.png) [@oripwk](https://discuss.kotlinlang.org/u/oripwk)\
**Post date:** [April 18, 2018, 6:04pm UTC](https://discuss.kotlinlang.org/t/creating-a-channel-from-exhaustible-source/7464/1 "2018-04-18T18:04:26Z")

</div>

Hi,

Suppose I have some exhaustible source (such as an SQS queue) from which I want to create a channel. What would be the correct strategy to implement such thing?

My draft implementation looks like this:

```
fun sqsChannel(queueUrl: String, interval: Long, sqs: AmazonSQSAsyncClient): ReceiveChannel<String> =
    produce {
        while (true) {
            val result = sqs.receiveMessage(queueUrl) // assume receiveMessage() is a suspend function
            if (result.messages.isEmpty()) {
                delay(interval)
            } else {
                result.messages.forEach { send(it.body) }
            }
        }
    }

```

Is the implementation correct? (i.e. won’t cause deadlocks, will propagate backpressure, etc.)

Thanks

---

<div class="post-metadata">

**Author:** ![noo.blaster](https://avatars.discourse-cdn.com/v4/letter/n/ac91a4/32.png) [@noo.blaster](https://discuss.kotlinlang.org/u/noo.blaster)\
**Post date:** [April 20, 2018, 9:25am UTC](https://discuss.kotlinlang.org/t/creating-a-channel-from-exhaustible-source/7464/2 "2018-04-20T09:25:56Z")

</div>

I don’t see any reason why this would cause a deadlock in your current code.

the channel (here without buffer… maybe you should add capacity to your produce) is handling back pressure (send will potentially suspend current coroutine).

but something looks bad to me : `delay(interval)`. you are actually polling an asyncClient.  
I don’t know anything about AmazonSQSAsyncClient but maybe you should use [an async method](https://github.com/aws/aws-sdk-java/blob/master/aws-java-sdk-sqs/src/main/java/com/amazonaws/services/sqs/AmazonSQSAsyncClient.java#L510) combined with [suspend-coroutine](https://kotlinlang.org/api/latest/jvm/stdlib/kotlin.coroutines.experimental/suspend-coroutine.html) to let AmazonSQSAsyncClient handle the async stuff.

to go further, internally AmazonSQSAsyncClient is using an `executorService` of blocking thread: this is not really that good… so if you feel adventurous, you can reimplement a real async client.

---

<div class="post-metadata">

**Author:** ![oripwk](https://sea1.discourse-cdn.com/flex019/user_avatar/discuss.kotlinlang.org/oripwk/32/5028_2.png) [@oripwk](https://discuss.kotlinlang.org/u/oripwk)\
**Post date:** [April 21, 2018, 10:11am UTC](https://discuss.kotlinlang.org/t/creating-a-channel-from-exhaustible-source/7464/3 "2018-04-21T10:11:12Z")

</div>

Thanks for the helpful response.

The code I provided was deliberately simplified. `receiveMessage()` is actually wrapped in a method that runs the call on a separate worker thread pool and returns a coroutine that doesn’t block the current thread pool.

As for implementing an async client, it’s already a [work in progress](https://aws.amazon.com/blogs/developer/aws-sdk-for-java-2-0-developer-preview/).