/*
 * Copyright 2014-2018 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.cluster.service;

import io.aeron.Aeron;
import io.aeron.Publication;
import org.agrona.CloseHelper;
import org.agrona.DirectBuffer;

/**
 * Session representing a connected client to the cluster.
 */
public class ClientSession
{
    /**
     * Return value to indicate egress to a session is mocked out by the cluster when in follower mode.
     */
    public static final long MOCKED_OFFER = 1;

    private final long id;
    private final int responseStreamId;
    private final String responseChannel;
    private final byte[] encodedPrincipal;

    private final ClusteredServiceAgent cluster;
    private Publication responsePublication;
    private boolean isClosing;

    ClientSession(
        final long sessionId,
        final int responseStreamId,
        final String responseChannel,
        final byte[] encodedPrincipal,
        final ClusteredServiceAgent cluster)
    {
        this.id = sessionId;
        this.responseStreamId = responseStreamId;
        this.responseChannel = responseChannel;
        this.encodedPrincipal = encodedPrincipal;
        this.cluster = cluster;
    }

    /**
     * Cluster session identity uniquely allocated when the session was opened.
     *
     * @return the cluster session identity uniquely allocated when the session was opened.
     */
    public long id()
    {
        return id;
    }

    /**
     * The response channel stream id for responding to the client.
     *
     * @return response channel stream id for responding to the client.
     */
    public int responseStreamId()
    {
        return responseStreamId;
    }

    /**
     * The response channel for responding to the client.
     *
     * @return response channel for responding to the client.
     */
    public String responseChannel()
    {
        return responseChannel;
    }

    /**
     * Cluster session encoded principal from when the session was authenticated.
     *
     * @return The encoded Principal passed. May be 0 length to indicate none present.
     */
    public byte[] encodedPrincipal()
    {
        return encodedPrincipal;
    }

    /**
     * Request the close of this {@link ClientSession} by sending the request to the consensus module.
     *
     * @return true if the event to close a session was sent or false if back pressure was applied.
     */
    public boolean close()
    {
        try
        {
            return cluster.closeSession(id);
        }
        catch (final IllegalArgumentException ignored)
        {
            return true;
        }
    }

    /**
     * Indicates that a request to close this session has been made.
     *
     * @return whether a request to close this session has been made.
     */
    public boolean isClosing()
    {
        return isClosing;
    }

    /**
     * Non-blocking publish of a partial buffer containing a message to a cluster.
     *
     * @param buffer containing message.
     * @param offset offset in the buffer at which the encoded message begins.
     * @param length in bytes of the encoded message.
     * @return the same as {@link Publication#offer(DirectBuffer, int, int)} when in {@link Cluster.Role#LEADER}
     * otherwise {@link #MOCKED_OFFER}.
     */
    public long offer(final DirectBuffer buffer, final int offset, final int length)
    {
        return cluster.offer(id, responsePublication, buffer, offset, length);
    }

    void connect(final Aeron aeron)
    {
        if (null == responsePublication)
        {
            responsePublication = aeron.addExclusivePublication(responseChannel, responseStreamId);
        }
    }

    void markClosing()
    {
        this.isClosing = true;
    }

    void resetClosing()
    {
        isClosing = false;
    }

    void disconnect()
    {
        CloseHelper.close(responsePublication);
        responsePublication = null;
    }
}

@@ -2725,21 +2725,9 @@
   * 
-Request the c
+C
 lose
@@ -2807,32 +2807,36 @@
 dule.%0A     *
+ %3Cp%3E
 %0A     * 
 @return true
@@ -2827,99 +2827,33 @@
   * 
-@return true if the event to close a session was sent or false if back pressure was applied
+This method is idempotent
 .%0A  
@@ -2869,23 +2869,20 @@
  public 
-boolean
+void
  close()
@@ -2900,54 +2900,37 @@
     
-try%0A        %7B%0A            return cluster.close
+if (null != cluster.getClient
 Sess
@@ -2936,17 +2936,17 @@
 sion(id)
-;
+)
 %0A       
@@ -2942,25 +2942,25 @@
 d))%0A        
-%7D
+%7B
 %0A        cat
@@ -2960,88 +2960,36 @@
     
-catch (final IllegalArgumentException ignored)%0A        %7B%0A            return true
+    cluster.closeSession(id)
 ;%0A  

Make cluster ClientSession.close() consistent with other close methods.