protected def runKinesisSourceWithInspection()

in src/it/scala/com/gu/kinesis/KinesisTestComponents.scala [75:82]

• 8 lines of code
• 1 McCabe index (conditional complexity)

  protected def runKinesisSourceWithInspection(
      kinesisSource: Source[KinesisRecord, NotUsed]
  ): () => IndexedSeq[KeyAndMessage] = {
    kinesisSource
      .via(markRecordsAsProcessed)
      .via(extractKeyAndMessage)
      .runWith(Inspectable.sink)
  }