/*
 * Copyright 2014 - 2016 Real Logic Ltd.
 *
 * 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.aeron;

import io.aeron.command.CorrelatedMessageFlyweight;
import io.aeron.command.PublicationMessageFlyweight;
import io.aeron.command.RemoveMessageFlyweight;
import io.aeron.command.SubscriptionMessageFlyweight;
import org.agrona.concurrent.UnsafeBuffer;
import org.agrona.concurrent.ringbuffer.RingBuffer;

import java.nio.ByteBuffer;

import static io.aeron.command.ControlProtocolEvents.*;

/**
 * Separates the concern of communicating with the client conductor away from the rest of the client.
 *
 * Writes messages into the client conductor buffer.
 */
public class DriverProxy
{
    /** Maximum capacity of the write buffer */
    public static final int MSG_BUFFER_CAPACITY = 4096;

    private final long clientId;
    private final UnsafeBuffer buffer = new UnsafeBuffer(ByteBuffer.allocateDirect(MSG_BUFFER_CAPACITY));
    private final PublicationMessageFlyweight publicationMessage = new PublicationMessageFlyweight();
    private final SubscriptionMessageFlyweight subscriptionMessage = new SubscriptionMessageFlyweight();

    private final RemoveMessageFlyweight removeMessage = new RemoveMessageFlyweight();
    // the heartbeats come from the client conductor thread, so keep the flyweights and buffer separate
    private final UnsafeBuffer keepaliveBuffer = new UnsafeBuffer(ByteBuffer.allocateDirect(MSG_BUFFER_CAPACITY));

    private final CorrelatedMessageFlyweight correlatedMessage = new CorrelatedMessageFlyweight();
    private final RingBuffer toDriverCommandBuffer;

    public DriverProxy(final RingBuffer toDriverCommandBuffer)
    {
        this.toDriverCommandBuffer = toDriverCommandBuffer;

        publicationMessage.wrap(buffer, 0);
        subscriptionMessage.wrap(buffer, 0);

        correlatedMessage.wrap(keepaliveBuffer, 0);
        removeMessage.wrap(buffer, 0);

        clientId = toDriverCommandBuffer.nextCorrelationId();
    }

    public long timeOfLastDriverKeepalive()
    {
        return toDriverCommandBuffer.consumerHeartbeatTime();
    }

    public long addPublication(final String channel, final int streamId)
    {
        final long correlationId = toDriverCommandBuffer.nextCorrelationId();

        publicationMessage
            .clientId(clientId)
            .correlationId(correlationId);

        publicationMessage
            .streamId(streamId)
            .channel(channel);

        if (!toDriverCommandBuffer.write(ADD_PUBLICATION, buffer, 0, publicationMessage.length()))
        {
            throw new IllegalStateException("could not write publication message");
        }

        return correlationId;
    }

    public long removePublication(final long registrationId)
    {
        final long correlationId = toDriverCommandBuffer.nextCorrelationId();
        removeMessage.correlationId(correlationId);
        removeMessage.registrationId(registrationId);

        if (!toDriverCommandBuffer.write(REMOVE_PUBLICATION, buffer, 0, RemoveMessageFlyweight.length()))
        {
            throw new IllegalStateException("could not write publication remove message");
        }

        return correlationId;
    }

    public long addSubscription(final String channel, final int streamId)
    {
        final long registrationId = -1;
        final long correlationId = toDriverCommandBuffer.nextCorrelationId();

        subscriptionMessage
            .clientId(clientId)
            .correlationId(correlationId);

        subscriptionMessage
            .registrationCorrelationId(registrationId)
            .streamId(streamId)
            .channel(channel);

        if (!toDriverCommandBuffer.write(ADD_SUBSCRIPTION, buffer, 0, subscriptionMessage.length()))
        {
            throw new IllegalStateException("could not write subscription message");
        }

        return correlationId;
    }

    public long removeSubscription(final long registrationId)
    {
        final long correlationId = toDriverCommandBuffer.nextCorrelationId();
        removeMessage.correlationId(correlationId);
        removeMessage.registrationId(registrationId);

        if (!toDriverCommandBuffer.write(REMOVE_SUBSCRIPTION, buffer, 0, RemoveMessageFlyweight.length()))
        {
            throw new IllegalStateException("could not write subscription remove message");
        }

        return correlationId;
    }

    public void sendClientKeepalive()
    {
        correlatedMessage
            .clientId(clientId)
            .correlationId(0);

        if (!toDriverCommandBuffer.write(CLIENT_KEEPALIVE, keepaliveBuffer, 0, CorrelatedMessageFlyweight.LENGTH))
        {
            throw new IllegalStateException("could not write keepalive message");
        }
    }
}

@@ -1165,16 +1165,133 @@
 buffer.%0A
+ *%0A * Note: this class is not thread safe and is expecting to be called under the %7B@link ClientConductor%7D main lock.%0A
  */%0Apubl
@@ -1422,41 +1422,8 @@
 6;%0A%0A
-    private final long clientId;%0A
     
@@ -1823,228 +1823,8 @@
 ();%0A
-    // the heartbeats come from the client conductor thread, so keep the flyweights and buffer separate%0A    private final UnsafeBuffer keepaliveBuffer = new UnsafeBuffer(ByteBuffer.allocateDirect(MSG_BUFFER_CAPACITY));%0A%0A
     
@@ -2182,33 +2182,32 @@
 rap(buffer, 0);%0A
-%0A
         correlat
@@ -2221,26 +2221,17 @@
 ge.wrap(
-keepaliveB
+b
 uffer, 0
@@ -2281,16 +2281,27 @@
         
+final long 
 clientId
@@ -2342,24 +2342,70 @@
 lationId();%0A
+        correlatedMessage.clientId(clientId);%0A
     %7D%0A%0A    p
@@ -2706,53 +2706,8 @@
 sage
-%0A            .clientId(clientId)%0A            
 .cor
@@ -3228,39 +3228,53 @@
 emoveMessage
-.correl
+%0A            .registr
 ationId(
 correlationI
@@ -3257,38 +3257,39 @@
 istrationId(
-correl
+registr
 ationId)
 ;%0A        re
@@ -3272,34 +3272,35 @@
 istrationId)
-;
 %0A
+  
         
 removeMessag
@@ -3291,37 +3291,25 @@
     
-removeMessage.registr
+  .correl
 ationId(
 regi
@@ -3296,39 +3296,38 @@
  .correlationId(
-registr
+correl
 ationId);%0A%0A     
@@ -3806,53 +3806,8 @@
 sage
-%0A            .clientId(clientId)%0A            
 .cor
@@ -4388,39 +4388,53 @@
 emoveMessage
-.correl
+%0A            .registr
 ationId(
 correlationI
@@ -4421,30 +4421,31 @@
 ationId(
-correl
+registr
 ationId)
 ;%0A      
@@ -4436,26 +4436,27 @@
 ationId)
-;
 %0A
+  
         
 removeMe
@@ -4451,37 +4451,25 @@
     
-removeMessage.registr
+  .correl
 ationId(
 regi
@@ -4456,39 +4456,38 @@
  .correlationId(
-registr
+correl
 ationId);%0A%0A     
@@ -4811,53 +4811,8 @@
 sage
-%0A            .clientId(clientId)%0A            
 .cor
@@ -4890,18 +4890,9 @@
 VE, 
-keepaliveB
+b
 uffe

Simplify DriverProxy as it is only called on the conductor lock and benefit from the clientId field set in the buffer.