# Alpakka AMQP Consuming
**URL:** https://discuss.akka.io/t/alpakka-amqp-consuming/3560
**Category:** Akka Streams & Alpakka
**Created:** [February 26, 2019, 10:23pm UTC](https://discuss.akka.io/t/alpakka-amqp-consuming/3560 "2019-02-26T22:23:05Z")
**Posts on this page:** 5
**Page:** 1
### Author:  [@rincerwind](https://discuss.akka.io/u/rincerwind)
#### Post date: [February 26, 2019, 10:23pm UTC](https://discuss.akka.io/t/alpakka-amqp-consuming/3560/1 "2019-02-26T22:23:05Z")
Hello,
Akka-Streams are quite new to me and I wanted to get some hands-on experience. I was thinking of setting up a Qpid broker and building a Producer and Consumer for it.
While looking at Alpakka-AMQP, I noticed that there doesn’t seem to be (at first glance) a way to setup a listener, which will receive messages from the broker once those have been published by the Producer.
I can successfully run the example from the documentation but once I consume the messages, the connection gets closed and the consumer is no longer listening. Specifically, I refer to the example about receiving:
```auto
val amqpSource = AmqpSource.atMostOnceSource(
NamedQueueSourceSettings(connectionProvider, queueName).withDeclaration(queueDeclaration),
bufferSize = 10
)
val result = amqpSource.take(5).runWith(Sink.seq)
```
I am most likely missing something and will appreciate your help.
Thank you for your time!
---
### Author:  [@johanandren](https://discuss.akka.io/u/johanandren)
#### Post date: [February 27, 2019, 8:35am UTC](https://discuss.akka.io/t/alpakka-amqp-consuming/3560/2 "2019-02-27T08:35:39Z")
Hi Toni, the `take(n)` operator will accept at most `n` elements before it cancels upstream, so if you want it to continuously process elements from the topic you should use `take`. Additionally `Sink.seq` will collect all elements into memory so that would be a pretty bad idea if the stream is “infinite”.
The collect finite number of elements logic in the sample is likely to keep the sample as simple as possible.
Another basic example that will keep running “forever” (until you terminate the `ActorSystem` backing it), and print each element, would be:
```auto
Source(Nil).runWith(Sink.foreach(elem => println(elem)))
```
---
### Author:  [@rincerwind](https://discuss.akka.io/u/rincerwind)
#### Post date: [February 27, 2019, 11:17am UTC](https://discuss.akka.io/t/alpakka-amqp-consuming/3560/3 "2019-02-27T11:17:08Z")
Hi Johan,
Thank you for your reply.
I was unable to find `take` in the API (I am using akka-stream 2.5.16). The second suggestion worked though!
Would this be expensive - to keep the stream open forever?
---
### Author:  [@johanandren](https://discuss.akka.io/u/johanandren)
#### Post date: [February 27, 2019, 12:36pm UTC](https://discuss.akka.io/t/alpakka-amqp-consuming/3560/4 "2019-02-27T12:36:40Z")
I was referring to the sample code there, that uses the `take` operator in `amqpSource.take(5).runWith(Sink.seq)`, it is avalabile on both `Source` and `Flow`
In general I’d recommend to read up on the basic concepts, Akka Streams can be hard to trial-and-error-learn. It will likely save you time and grief. The quickstart in the docs for example: [https://doc.akka.io/docs/akka/current/stream/stream-quickstart.html](https://doc.akka.io/docs/akka/current/stream/stream-quickstart.html) (there are also two more learning paths in the docs [https://doc.akka.io/docs/akka/current/stream/stream-introduction.html#how-to-read-these-docs](https://doc.akka.io/docs/akka/current/stream/stream-introduction.html#how-to-read-these-docs) )
Running a stream “forever” is one of the most common use cases for using Akka Streams, so this is a fine idea.
---
### Author:  [@rincerwind](https://discuss.akka.io/u/rincerwind)
#### Post date: [February 28, 2019, 11:59am UTC](https://discuss.akka.io/t/alpakka-amqp-consuming/3560/5 "2019-02-28T11:59:28Z")
Thanks, Johan, I will go through the documentation.