/*
 * Copyright (C) 2015 SoftIndex LLC.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.datakernel.rpc.protocol.stream;

import io.datakernel.eventloop.AsyncTcpSocket;
import io.datakernel.eventloop.Eventloop;
import io.datakernel.rpc.protocol.RpcConnection;
import io.datakernel.rpc.protocol.RpcControlMessage;
import io.datakernel.rpc.protocol.RpcMessage;
import io.datakernel.rpc.protocol.RpcProtocol;
import io.datakernel.serializer.BufferSerializer;
import io.datakernel.stream.SimpleStreamConsumer;
import io.datakernel.stream.SimpleStreamProducer;
import io.datakernel.stream.StreamStatus;
import io.datakernel.stream.net.SocketStreamingConnection;
import io.datakernel.stream.processor.StreamBinaryDeserializer;
import io.datakernel.stream.processor.StreamBinarySerializer;
import io.datakernel.stream.processor.StreamLZ4Compressor;
import io.datakernel.stream.processor.StreamLZ4Decompressor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@SuppressWarnings("unchecked")
final class RpcStreamProtocol implements RpcProtocol {
	private final Logger logger = LoggerFactory.getLogger(this.getClass());

	private final RpcConnection rpcConnection;
	private final SimpleStreamProducer<RpcMessage> sender;
	private final SimpleStreamConsumer<RpcMessage> receiver;
	private final SocketStreamingConnection connection;

	protected RpcStreamProtocol(Eventloop eventloop, AsyncTcpSocket asyncTcpSocket,
	                            RpcConnection rpcConnection,
	                            BufferSerializer<RpcMessage> messageSerializer,
	                            int defaultPacketSize, int maxPacketSize, boolean compression, boolean server) {
		this.rpcConnection = rpcConnection;

		if (server) {
			sender = SimpleStreamProducer
					.create(eventloop)
					.withStatusListener(new SimpleStreamProducer.StatusListener() {
						@Override
						public void onResumed() {
							receiver.resume();
						}

						@Override
						public void onSuspended() {
							receiver.suspend();
						}

						@Override
						public void onClosedWithError(Exception e) {
							RpcStreamProtocol.this.rpcConnection.onClosedWithError(e);
						}
					});
		} else {
			sender = SimpleStreamProducer
					.create(eventloop)
					.withStatusListener(new SimpleStreamProducer.StatusListener() {
						@Override
						public void onResumed() {
						}

						@Override
						public void onSuspended() {
						}

						@Override
						public void onClosedWithError(Exception e) {
							RpcStreamProtocol.this.rpcConnection.onClosedWithError(e);
						}
					});
		}

		receiver = SimpleStreamConsumer.create(eventloop, new SimpleStreamConsumer.StatusListener() {
			@Override
			public void onEndOfStream() {
				RpcStreamProtocol.this.rpcConnection.onReadEndOfStream();
			}

			@Override
			public void onClosedWithError(Exception e) {
				RpcStreamProtocol.this.rpcConnection.onClosedWithError(e);
			}
		}, rpcConnection);

		StreamBinarySerializer<RpcMessage> serializer = StreamBinarySerializer.create(eventloop, messageSerializer, defaultPacketSize,
				maxPacketSize, 0, true);
		StreamBinaryDeserializer<RpcMessage> deserializer = StreamBinaryDeserializer.create(eventloop, messageSerializer, maxPacketSize);
		this.connection = SocketStreamingConnection.createSocketStreamingConnection(eventloop, asyncTcpSocket);

		if (compression) {
			StreamLZ4Compressor compressor = StreamLZ4Compressor.fastCompressor(eventloop);
			StreamLZ4Decompressor decompressor = StreamLZ4Decompressor.create(eventloop);
			connection.receiveStreamTo(decompressor.getInput());
			decompressor.getOutput().streamTo(deserializer.getInput());

			serializer.getOutput().streamTo(compressor.getInput());
			connection.sendStreamFrom(compressor.getOutput());
		} else {
			connection.receiveStreamTo(deserializer.getInput());
			connection.sendStreamFrom(serializer.getOutput());
		}

		deserializer.getOutput().streamTo(receiver);
		sender.streamTo(serializer.getInput());
	}

	@Override
	public void sendMessage(RpcMessage message) {
		sender.send(message);
	}

	@Override
	public void sendCloseMessage() {
		sender.send(RpcMessage.of(-1, RpcControlMessage.CLOSE));
	}

	@Override
	public boolean isOverloaded() {
		return sender.getProducerStatus() == StreamStatus.SUSPENDED;
	}

	@Override
	public AsyncTcpSocket.EventHandler getSocketConnection() {
		return connection;
	}

	@Override
	public void sendEndOfStream() {
		sender.sendEndOfStream();
	}
}

@@ -725,16 +725,64 @@
 ntloop;%0A
+import io.datakernel.exception.SimpleException;%0A
 import i
@@ -814,32 +814,32 @@
 .RpcConnection;%0A
-
 import io.datake
@@ -1609,16 +1609,144 @@
 tocol %7B%0A
+%09private static final SimpleException CONNECTION_CLOSED_EXCEPTION =%0A%09%09%09new SimpleException(%22Rpc %D1%81onnection is already closed%22);%0A
 %09private
@@ -1811,16 +1811,52 @@
 ss());%0A%0A
+%09private final Eventloop eventloop;%0A
 %09private
@@ -2275,16 +2275,16 @@
 alizer,%0A
-
 %09       
@@ -2385,16 +2385,46 @@
 rver) %7B%0A
+%09%09this.eventloop = eventloop;%0A
 %09%09this.r
@@ -4782,23 +4782,26 @@
 %7B%0A%09%09send
-er.send
+RpcMessage
 (message
@@ -4858,23 +4858,26 @@
 %7B%0A%09%09send
-er.send
+RpcMessage
 (RpcMess
@@ -4910,32 +4910,236 @@
 ge.CLOSE));%0A%09%7D%0A%0A
+%09private void sendRpcMessage(RpcMessage message) %7B%0A%09%09if (sender.getProducerStatus().isClosed()) %7B%0A%09%09%09eventloop.recordIoError(CONNECTION_CLOSED_EXCEPTION, this);%0A%09%09%09return;%0A%09%09%7D%0A%09%09sender.send(message);%0A%09%7D%0A%0A
 %09@Override%0A%09publ

handle unexpected connection closing in rpc