package cn.strong.fastdfs.client.handler;

import static cn.strong.fastdfs.client.CommandCodes.FDFS_PROTO_CMD_RESP;
import static java.nio.charset.StandardCharsets.UTF_8;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.embedded.EmbeddedChannel;

import java.nio.charset.Charset;

import org.junit.Test;

import cn.strong.fastdfs.client.Consts;
import cn.strong.fastdfs.client.protocol.response.AbstractReceiver;
import cn.strong.fastdfs.client.protocol.response.Receiver;
import cn.strong.fastdfs.utils.Utils;

public class FastdfsDecoderTest {

	@Test
	public void test() {
		StubReceiver receiver = new StubReceiver();
		receiver.observable().subscribe(txt -> {
			System.out.println("response: " + txt);
		}, ex -> {
			ex.printStackTrace();
		}, () -> {
			System.out.println("completed");
		});

		String msg = "Hello world!";
		byte[] bytes = msg.getBytes();
		ByteBuf buf = Unpooled.buffer();
		buf.writeLong(bytes.length);
		buf.writeByte(FDFS_PROTO_CMD_RESP);
		buf.writeByte(Consts.ERRNO_OK);
		buf.writeBytes(bytes);

		FastdfsDecoder handler = new FastdfsDecoder();
		EmbeddedChannel channel = new EmbeddedChannel(handler);
		channel.attr(Receiver.RECEIVER).set(receiver);
		channel.attr(Consts.CHARSET).set(UTF_8);

		// write bytes
		channel.writeInbound(buf);
		channel.finish();
	}

	private static class StubReceiver extends AbstractReceiver<String> {
		private long length;

		@Override
		public void setLength(long length) {
			this.length = length;
		}

		@Override
		public boolean tryRead(ByteBuf in, Charset charset) {
			String txt = Utils.readString(in, (int) length, charset);
			subject.onNext(txt);
			subject.onCompleted();
			return true;
		}
	}
}

@@ -167,123 +167,8 @@
 F_8;
-%0Aimport io.netty.buffer.ByteBuf;%0Aimport io.netty.buffer.Unpooled;%0Aimport io.netty.channel.embedded.EmbeddedChannel;
 %0A%0Aim
@@ -428,16 +428,131 @@
 s.Utils;
+%0Aimport io.netty.buffer.ByteBuf;%0Aimport io.netty.buffer.Unpooled;%0Aimport io.netty.channel.embedded.EmbeddedChannel;
 %0A%0Apublic
@@ -670,96 +670,69 @@
 ver.
-observable().subscribe(txt -%3E %7B%0A%09%09%09System.out.println(%22response: %22 + txt);%0A%09%09%7D, ex -%3E
+promise().whenComplete((data, ex) -%3E %7B%0A%09%09%09if (ex != null)
  %7B%0A
+%09
 %09%09%09e
@@ -758,19 +758,19 @@
 ;%0A%09%09
-%7D, () -%3E
+%09%7D else
  %7B%0A
+%09
 %09%09%09S
@@ -792,20 +792,33 @@
 ln(%22
-completed%22);
+response: %22 + data);%0A%09%09%09%7D
 %0A%09%09%7D
@@ -1630,52 +1630,28 @@
 %0A%09%09%09
-subject.onNext(txt);%0A%09%09%09subject.onC
+promise.c
 omplete
-d
 (
+txt
 );%0A%09

update FastdfsDecoderTest