package com.emc.nautilus.logclient.impl;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

import org.junit.Test;

import com.emc.nautilus.common.netty.ClientConnection;
import com.emc.nautilus.common.netty.ConnectionFactory;
import com.emc.nautilus.common.netty.ConnectionFailedException;
import com.emc.nautilus.common.netty.ReplyProcessor;
import com.emc.nautilus.common.netty.WireCommands.AppendData;
import com.emc.nautilus.common.netty.WireCommands.AppendSetup;
import com.emc.nautilus.common.netty.WireCommands.SetupAppend;
import com.emc.nautilus.logclient.SegmentSealedExcepetion;
import com.emc.nautilus.logclient.impl.SegmentOutputStreamImpl;

import static org.junit.Assert.*;

import static org.mockito.Mockito.*;

import io.netty.buffer.Unpooled;
import lombok.Synchronized;

public class SegmentOutputStreamTest {

    private static class TestConnectionFactoryImpl implements ConnectionFactory {
        Map<String, ClientConnection> connections = new HashMap<>();
        Map<String, ReplyProcessor> processors = new HashMap<>();

        @Override
        @Synchronized
        public ClientConnection establishConnection(String endpoint, ReplyProcessor rp) {
            ClientConnection connection = connections.get(endpoint);
            if (connection == null) {
                throw new IllegalStateException("Unexpected Endpoint");
            }
            processors.put(endpoint, rp);
            return connection;
        }

        @Synchronized
        void provideConnection(String endpoint, ClientConnection c) {
            connections.put(endpoint, c);
        }

        @Synchronized
        ReplyProcessor getProcessor(String endpoint) {
            return processors.get(endpoint);
        }

        @Override
        public void shutdown() {
        }
    }

    private static ByteBuffer getBuffer(String s) {
        return ByteBuffer.wrap(s.getBytes());
    }

    @Test
    public void testConnectAndSend() throws SegmentSealedExcepetion, ConnectionFailedException {
        UUID cid = UUID.randomUUID();
        TestConnectionFactoryImpl cf = new TestConnectionFactoryImpl();
        ClientConnection connection = mock(ClientConnection.class);
        cf.provideConnection("endpoint", connection);  
        SegmentOutputStreamImpl output = new SegmentOutputStreamImpl(cf, "endpoint", cid, "segment");
        output.connect();
        verify(connection).send(new SetupAppend(cid, "segment"));
        cf.getProcessor("endpoint").appendSetup(new AppendSetup("segment", cid, 0));
        
        CompletableFuture<Void> acked = new CompletableFuture<>();
        ByteBuffer data = getBuffer("test");
        output.write(data, acked);
        verify(connection).send(new AppendData(cid, data.array().length, Unpooled.wrappedBuffer(data)));
        verifyNoMoreInteractions(connection);
        assertEquals(false, acked.isDone());
    }

    @Test
    public void testNewEventsGoAfterInflight() {
        fail();
    }

    @Test
    public void testClose() {
        fail();
    }

    @Test
    public void testFlush() {
        fail();
    }

    @Test
    public void testAutoClose() {
        fail();
    }

    @Test
    public void testFailOnAutoClose() {
        fail();
    }

    @Test
    public void testOutOfOrderAcks() {
        fail();
    }

    @Test
    public void testLargeWrite() {
        fail();
    }
}

@@ -2638,144 +2638,581 @@
     
-CompletableFuture%3CVoid%3E acked = new CompletableFuture%3C%3E();%0A        ByteBuffer data = getBuffer(%22test%22);%0A        output.write(data, acked
+sendEvent(cid, connection, output, %22test%22);%0A        verifyNoMoreInteractions(connection);%0A    %7D%0A%0A    @Test%0A    public void testNewEventsGoAfterInflight() throws ConnectionFailedException, SegmentSealedExcepetion %7B%0A        UUID cid = UUID.randomUUID();%0A        TestConnectionFactoryImpl cf = new TestConnectionFactoryImpl();%0A        ClientConnection connection = mock(ClientConnection.class);%0A        cf.provideConnection(%22endpoint%22, connection);  %0A        SegmentOutputStreamImpl output = new SegmentOutputStreamImpl(cf, %22endpoint%22, cid, %22segment%22);%0A        output.connect(
 );%0A 
@@ -3250,74 +3250,211 @@
 new 
+Setup
 Append
-Data
 (cid, 
-data.array().length, Unpooled.wrappedBuffer(data))
+%22segment%22));%0A        cf.getProcessor(%22endpoint%22).appendSetup(new AppendSetup(%22segment%22, cid, 0));%0A        %0A        String event = %22test%22;%0A        sendEvent(cid, connection, output, event
 );%0A 
@@ -3506,128 +3506,488 @@
     
+%7D%0A%0A
     
-assertEquals(false, acked.isDone()
+private void sendEvent(UUID cid, ClientConnection connection, SegmentOutputStreamImpl output, String event)%0A            throws SegmentSealedExcepetion, ConnectionFailedException %7B%0A        CompletableFuture%3CVoid%3E acked = new CompletableFuture%3C%3E(
 );%0A    
-%7D%0A%0A
     
-@Test%0A    public void testNewEventsGoAfterInflight() %7B%0A        fail(
+ByteBuffer data = getBuffer(event);%0A        output.write(data, acked);%0A        verify(connection).send(new AppendData(cid, data.array().length, Unpooled.wrappedBuffer(data)));%0A        assertEquals(false, acked.isDone()
 );%0A 

Test update