package com.softwaremill.react.kafka

import kafka.consumer.KafkaConsumer
import org.reactivestreams.{Subscription, Subscriber}

import scala.util.control.NonFatal

private[kafka] class KafkaTopicSubscription(consumer: KafkaConsumer,
                                            subscriber: Subscriber[_ >: String]) extends Subscription {

  override def request(n: Long): Unit = {
    require(n > 0, "n <= 0")

    (1L to n.toInt) foreach { _ =>
      consumer.read(bytes => {
        val msgAsStr = new String(bytes)
        subscriber.onNext(msgAsStr)
      })
    }
  }

  override def cancel() = try {
    consumer.close()
  } catch {
    case NonFatal(exception) =>
      subscriber.onError(new IllegalStateException("onComplete threw an exception", exception))
  }
}

package com.softwaremill.react.kafka

import com.softwaremill.react.kafka.RichKafkaConsumer._
import kafka.consumer.KafkaConsumer
import org.reactivestreams.{Subscriber, Subscription}

import scala.util.control.NonFatal
private[kafka] class KafkaTopicSubscription(consumer: KafkaConsumer,
                                            subscriber: Subscriber[_ >: String]) extends Subscription {

  override def request(n: Long) {
    require(n > 0, "Rule 3.9: n <= 0")
    if (consumer.connected()) {
      (1L to n.toInt) foreach { _ =>
        consumer.read(bytes => {
          val msgAsStr = new String(bytes)
          subscriber.onNext(msgAsStr)
        })
      }
    } // else 3.6 NOP
  }

  override def cancel() = try {
    consumer.close()
  } catch {
    case NonFatal(exception) =>
      subscriber.onError(new IllegalStateException("onComplete threw an exception", exception))
  }
}

Implement rule 3.6 - After the Subscription is cancelled, additional Subscription.request(long n) MUST be NOPs
Implement rule 3.6
- After the Subscription is cancelled, additional Subscription.request(long n) MUST be NOPs
