# How to exit stream after n elements recieved?

**URL:** https://discuss.akka.io/t/how-to-exit-stream-after-n-elements-recieved/7252
**Category:** Akka Streams & Alpakka
**Created:** [September 26, 2020, 12:15am UTC](https://discuss.akka.io/t/how-to-exit-stream-after-n-elements-recieved/7252 "2020-09-26T00:15:46Z")
**Posts on this page:** 4
**Page:** 1

<div class="post-metadata">

### Author: ![jjones](https://avatars.discourse-cdn.com/v4/letter/j/df705f/32.png) [@jjones](https://discuss.akka.io/u/jjones)
#### Post date: [September 26, 2020, 12:15am UTC](https://discuss.akka.io/t/how-to-exit-stream-after-n-elements-recieved/7252/1 "2020-09-26T00:15:46Z")

</div>

Hello, I’m brand new to Akka and I’m just trying to get the hang of it.

As an experiment, I want to read from a Kinesis stream and collect n messages and stop.

The only one I found that would stop reading records was Sink.head(). But that only returns one record, I’d like to get more than that.

I can’t quite figure out how to stop reading from the stream after receiving the n messages though.

Here’s the code I have tried so far

```auto
  @Test
  public void testReadingFromKinesisNRecords() throws ExecutionException, InterruptedException {
    final ActorSystem system = ActorSystem.create("foo");
    final Materializer materializer = ActorMaterializer.create(system);

    ProfileCredentialsProvider profileCredentialsProvider = ProfileCredentialsProvider.create();

    final KinesisAsyncClient kinesisClient = KinesisAsyncClient.builder()
        .credentialsProvider(profileCredentialsProvider)
        .region(Region.US_WEST_2)
            .httpClient(AkkaHttpClient.builder()
                .withActorSystem(system).build())
            .build();

    system.registerOnTermination(kinesisClient::close);

    String streamName = "akka-test-stream";
    String shardId = "shardId-000000000000";

    int numberOfRecordsToRead = 3;

    final ShardSettings settings = ShardSettings.create(streamName, shardId)
            .withRefreshInterval(Duration.ofSeconds(1))
            .withLimit(numberOfRecordsToRead) // return a maximum of n records (and quit?!)
            .withShardIterator(ShardIterators.latest());

    final Source<Record, NotUsed> sourceKinesisBasic = KinesisSource.basic(settings, kinesisClient);

    Flow<Record, String, NotUsed> flowMapRecordToString = Flow.of(Record.class).map(record -> extractDataFromRecord(record));
    Flow<String, String, NotUsed> flowPrinter = Flow.of(String.class).map(s -> debugPrint(s));
// Flow<String, List<String>, NotUsed> flowGroupedWithinMinute =
// Flow.of(String.class).groupedWithin(
// numberOfRecordsToRead, // group size
// Duration.ofSeconds(60) // group time
// );

    Source<String, NotUsed> sourceStringsFromKinesisRecords = sourceKinesisBasic
        .via(flowMapRecordToString)
        .via(flowPrinter);
// .via(flowGroupedWithinMinute); // nope

    // sink to list of strings
// Sink<String, CompletionStage<List<String>>> sinkToList = Sink.seq();
    Sink<String, CompletionStage<List<String>>> sink10 = Sink.takeLast(10);
// Sink<String, CompletionStage<String>> sinkHead = Sink.head(); // only gives you one message

    CompletionStage<List<String>> streamCompletion = sourceStringsFromKinesisRecords
        .runWith(sink10, materializer);
    CompletableFuture<List<String>> completableFuture = streamCompletion.toCompletableFuture();
    completableFuture.join(); // never stops running...
    List<String> result = completableFuture.get();
    int foo = 1;
  }

  private String extractDataFromRecord(Record record) {
    String encType = record.encryptionTypeAsString();
    Instant arrivalTimestamp = record.approximateArrivalTimestamp();
    String data = record.data().asString(StandardCharsets.UTF_8);
    return data;
  }

  private String debugPrint(String s) {
    System.out.println(s);
    return s;
  }

```

Thank you for any clues

---

<div class="post-metadata">

### Author: ![ennru](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/ennru/32/1495_2.png) [@ennru](https://discuss.akka.io/u/ennru)
#### Post date: [September 26, 2020, 4:58pm UTC](https://discuss.akka.io/t/how-to-exit-stream-after-n-elements-recieved/7252/2 "2020-09-26T16:58:14Z")

</div>

Hi Julie,

Welcome to Akka Streams!

It’s the `take` operator you are looking for.

I know it may be overwhelming, but skimming through the operators listing at [https://doc.akka.io/docs/akka/current/stream/operators/](https://doc.akka.io/docs/akka/current/stream/operators/) may help you to find other potentially helpful operators.

The [Streams Cookbook](https://doc.akka.io/docs/akka/current/stream/stream-cookbook.html) is another good point to learn more.

Cheers,  
Enno.

---

<div class="post-metadata">

### Author: ![ignatius](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/ignatius/32/1787_2.png) [@ignatius](https://discuss.akka.io/u/ignatius)
#### Post date: [September 27, 2020, 4:21pm UTC](https://discuss.akka.io/t/how-to-exit-stream-after-n-elements-recieved/7252/3 "2020-09-27T16:21:21Z")

</div>

To add to @ennru’s response, note that `take` is something that you do on the Flow level, not the Sink level like `Sink.head`. After `take(n)`, You can then do something with each element using another Sink, such as `Sink.foreach`.

---

<div class="post-metadata">

### Author: ![jjones](https://avatars.discourse-cdn.com/v4/letter/j/df705f/32.png) [@jjones](https://discuss.akka.io/u/jjones)
#### Post date: [September 27, 2020, 6:42pm UTC](https://discuss.akka.io/t/how-to-exit-stream-after-n-elements-recieved/7252/4 "2020-09-27T18:42:10Z")

</div>

Thanks it works now!

```auto
...
    Flow<String, String, NotUsed> flowTakeN = Flow.of(String.class).take(numberOfRecordsToRead);

    Source<String, NotUsed> sourceStringsFromKinesisRecords = sourceKinesisBasic
        .via(flowMapRecordToString)
        .via(flowPrinter)
        .via(flowTakeN);
...

```