@@ -0,0 +1,1693 @@
+package reactor.kafka;%0A%0Aimport org.reactivestreams.Subscriber;%0Aimport org.reactivestreams.Subscription;%0Aimport reactor.fn.Consumer;%0A%0Aimport java.util.concurrent.atomic.AtomicLong;%0Aimport java.util.function.LongBinaryOperator;%0A%0Apublic class KafkaSubscription%3CT%3E implements Subscription %7B%0A%0A  private final AtomicLong                     requested;%0A  private final Subscriber%3CT%3E                  subscriber;%0A  private final Consumer%3CKafkaSubscription%3CT%3E%3E cancel;%0A%0A  public KafkaSubscription(Subscriber%3CT%3E subscriber,%0A                           Consumer%3CKafkaSubscription%3CT%3E%3E cancel) %7B%0A    this.requested = new AtomicLong(0);%0A    this.subscriber = subscriber;%0A    this.cancel = cancel;%0A  %7D%0A%0A  @Override%0A  public void request(long l) %7B%0A    if (l %3C 1) %7B%0A      throw new RuntimeException(%22Can't request a non-positive number%22);%0A    %7D%0A%0A    requested.accumulateAndGet(l, new LongBinaryOperator() %7B%0A      @Override%0A      public long applyAsLong(long left, long right) %7B%0A        long sum = left + right;%0A        if (sum %3E= Long.MAX_VALUE) %7B%0A          return Long.MAX_VALUE; // Effectively unbounded%0A        %7D else %7B%0A          return sum;%0A        %7D%0A      %7D%0A    %7D);%0A  %7D%0A%0A  public void offer(T t) %7B%0A    long r = requested.accumulateAndGet(-1, new LongBinaryOperator() %7B%0A      @Override%0A      public long applyAsLong(long old, long diff) %7B%0A        long sum = old + diff;%0A%0A        if (sum %3E= Long.MAX_VALUE) %7B%0A          return Long.MAX_VALUE; // Effectively unbounded%0A        %7D else if (sum %3C= 0)%7B%0A          return 0;%0A        %7D else %7B%0A          return sum;%0A        %7D%0A      %7D%0A    %7D);%0A    if (r %3E= 0) %7B%0A      subscriber.onNext(t);%0A    %7D%0A  %7D%0A%0A%0A  @Override%0A  public void cancel() %7B%0A    cancel.accept(this);%0A  %7D%0A%7D%0A

Implement kafka subscription