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))
  }
}

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))
  }
}

--- 
+++ 
@@ -1,22 +1,23 @@
 package com.softwaremill.react.kafka
 
+import com.softwaremill.react.kafka.RichKafkaConsumer._
 import kafka.consumer.KafkaConsumer
-import org.reactivestreams.{Subscription, Subscriber}
+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): Unit = {
-    require(n > 0, "n <= 0")
-
-    (1L to n.toInt) foreach { _ =>
-      consumer.read(bytes => {
-        val msgAsStr = new String(bytes)
-        subscriber.onNext(msgAsStr)
-      })
-    }
+  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 {
Implement rule 3.6
- After the Subscription is cancelled, additional Subscription.request(long n) MUST be NOPs
