package com.softwaremill.react.kafka

import java.util.concurrent.atomic.AtomicReference

import kafka.producer.KafkaProducer
import org.reactivestreams.{Subscription, Subscriber}

private[kafka] class ReactiveKafkaSubscriber(val producer: KafkaProducer) extends Subscriber[String] {

  val active = new AtomicReference[Subscription]()

  override def onSubscribe(subscription: Subscription) {
    active.compareAndSet(null, subscription) match {
      case true => subscription.request(1)
      case false => subscription.cancel()
    }
  }

  override def onError(throwable: Throwable): Unit = {
    active.get().cancel()
  }

  override def onComplete() {
    active.get().cancel()
  }

  override def onNext(msg: String) {
    producer.send(msg)
    active.get().request(1)
  }

}

package com.softwaremill.react.kafka

import java.util.concurrent.atomic.AtomicReference

import kafka.producer.KafkaProducer
import org.reactivestreams.{Subscription, Subscriber}

private[kafka] class ReactiveKafkaSubscriber(val producer: KafkaProducer) extends Subscriber[String] {

  val active = new AtomicReference[Subscription]()

  override def onSubscribe(subscription: Subscription) {
    active.compareAndSet(null, subscription) match {
      case true => subscription.request(1)
      case false => subscription.cancel()
    }
  }

  override def onError(throwable: Throwable): Unit = {
    // TODO ?
  }

  override def onComplete() {
    // TODO ?
  }

  override def onNext(msg: String) {
    producer.send(msg)
    active.get().request(1)
  }

}

Revert "Cancel the subscription after onError / onComplete"
Revert "Cancel the subscription after onError / onComplete"

This reverts commit 855a703dfb0732948fc2c26bb79c060e799d7f26.
Subscription::cancel MUST NOT be called from Subscriber::onComplete (Rule 2.3)
