package de.openknowledge.reactive.charset;

import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;

import de.openknowledge.reactive.commons.AbstractSimpleProcessor;

public class DecodingProcessor extends AbstractSimpleProcessor<ByteBuffer, CharBuffer> {

  private CharsetDecoder decoder;
  private ByteBuffer byteBuffer;
  private CharBuffer charBuffer;

  public DecodingProcessor(Charset charset, int bufferSize) {
    decoder = charset.newDecoder();
    charBuffer = CharBuffer.allocate(bufferSize);
    charBuffer.mark();
  }

  @Override
  public void onNext(ByteBuffer buffer) {
    if (buffer == null) {
      throw new NullPointerException("buffer may not be null");
    }
    byteBuffer = buffer;
    charBuffer.reset();
    decoder.decode(buffer, charBuffer, false);
    if (charBuffer.position() == 0) {
      // TODO check if position 0 is equivalent to null
      super.request(1);
    } else {
      charBuffer.flip();
      charBuffer.mark();
      publish(charBuffer);
    }
  }

  @Override
  public void onComplete() {
    if (byteBuffer != null) {
      decoder.decode(byteBuffer, charBuffer, true);
      decoder.flush(charBuffer);
    }
    if (!charBuffer.hasRemaining()) {
      super.onComplete();
    } else if (isRequested()) {
      publish(charBuffer);
    } else {
      throw new IllegalStateException("Should not happen");
      // TODO what shall we do? We have remaining items but no one requested them, I think it should not happen
    }
  }
}

@@ -423,16 +423,45 @@
 rBuffer;
+%0A  private boolean completed;
 %0A%0A  publ
@@ -1388,188 +1388,232 @@
     
-%7D else %7B%0A      throw new IllegalStateException(%22Should not happen%22);%0A      // TODO what shall we do? We have remaining items but no one requested them, I think it should not happen
+  super.onComplete();%0A    %7D else %7B%0A      completed = true;%0A    %7D%0A  %7D%0A%0A  protected void request(long n) %7B%0A    if (completed && n %3E 0) %7B%0A      publish(charBuffer);%0A      super.onComplete();%0A    %7D else %7B%0A      super.request(n);
 %0A   

Support publishing the last chunk of characters delayed