/*
 *  Copyright 2017 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.client;

import io.aeron.ControlledFragmentAssembler;
import io.aeron.Subscription;
import io.aeron.cluster.codecs.MessageHeaderDecoder;
import io.aeron.cluster.codecs.NewLeaderEventDecoder;
import io.aeron.cluster.codecs.SessionEventDecoder;
import io.aeron.cluster.codecs.SessionHeaderDecoder;
import io.aeron.logbuffer.ControlledFragmentHandler;
import io.aeron.logbuffer.Header;
import org.agrona.DirectBuffer;

public class EgressPoller implements ControlledFragmentHandler
{
    private final int fragmentLimit;
    private final MessageHeaderDecoder messageHeaderDecoder = new MessageHeaderDecoder();
    private final SessionEventDecoder sessionEventDecoder = new SessionEventDecoder();
    private final NewLeaderEventDecoder newLeaderEventDecoder = new NewLeaderEventDecoder();
    private final SessionHeaderDecoder sessionHeaderDecoder = new SessionHeaderDecoder();
    private final ControlledFragmentAssembler fragmentAssembler = new ControlledFragmentAssembler(this);
    private final Subscription subscription;
    private long clusterSessionId = -1;
    private long correlationId = -1;
    private int templateId = -1;
    private boolean pollComplete = false;

    public EgressPoller(final Subscription subscription, final int fragmentLimit)
    {
        this.subscription = subscription;
        this.fragmentLimit = fragmentLimit;
    }

    /**
     * Get the {@link Subscription} used for polling events.
     *
     * @return the {@link Subscription} used for polling events.
     */
    public Subscription subscription()
    {
        return subscription;
    }

    /**
     * Cluster session id of the last polled event or -1 if poll returned nothing.
     *
     * @return cluster session id of the last polled event or -1 if unrecognised template.
     */
    public long clusterSessionId()
    {
        return clusterSessionId;
    }

    /**
     * Correlation id of the last polled event or -1 if poll returned nothing.
     *
     * @return correlation id of the last polled event or -1 if unrecognised template.
     */
    public long correlationId()
    {
        return correlationId;
    }

    /**
     * Has the last polling action received a complete event?
     *
     * @return true of the last polling action received a complete event?
     */
    public boolean isPollComplete()
    {
        return pollComplete;
    }

    /**
     * Get the template id of the last received event.
     *
     * @return the template id of the last received event.
     */
    public int templateId()
    {
        return templateId;
    }

    public int poll()
    {
        clusterSessionId = -1;
        correlationId = -1;
        templateId = -1;
        pollComplete = false;

        return subscription.controlledPoll(fragmentAssembler, fragmentLimit);
    }

    public ControlledFragmentAssembler.Action onFragment(
        final DirectBuffer buffer, final int offset, final int length, final Header header)
    {
        messageHeaderDecoder.wrap(buffer, offset);

        templateId = messageHeaderDecoder.templateId();
        switch (templateId)
        {
            case SessionEventDecoder.TEMPLATE_ID:
                sessionEventDecoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    messageHeaderDecoder.blockLength(),
                    messageHeaderDecoder.version());

                clusterSessionId = sessionEventDecoder.clusterSessionId();
                correlationId = sessionEventDecoder.correlationId();
                break;

            case NewLeaderEventDecoder.TEMPLATE_ID:
                newLeaderEventDecoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    messageHeaderDecoder.blockLength(),
                    messageHeaderDecoder.version());

                clusterSessionId = newLeaderEventDecoder.clusterSessionId();
                break;

            case SessionHeaderDecoder.TEMPLATE_ID:
                sessionHeaderDecoder.wrap(
                    buffer,
                    offset + MessageHeaderDecoder.ENCODED_LENGTH,
                    messageHeaderDecoder.blockLength(),
                    messageHeaderDecoder.version());

                clusterSessionId = sessionHeaderDecoder.clusterSessionId();
                correlationId = sessionHeaderDecoder.correlationId();
                break;

            default:
                throw new IllegalStateException("Unknown templateId: " + templateId);
        }

        pollComplete = true;

        return ControlledFragmentAssembler.Action.BREAK;
    }

    public MessageHeaderDecoder messageHeaderDecoder()
    {
        return messageHeaderDecoder;
    }

    public SessionEventDecoder sessionEventDecoder()
    {
        return sessionEventDecoder;
    }

    public SessionHeaderDecoder sessionHeaderDecoder()
    {
        return sessionHeaderDecoder;
    }

    public NewLeaderEventDecoder newLeaderEventDecoder()
    {
        return newLeaderEventDecoder;
    }
}

@@ -5267,427 +5267,7 @@
    %7D
-%0A%0A    public MessageHeaderDecoder messageHeaderDecoder()%0A    %7B%0A        return messageHeaderDecoder;%0A    %7D%0A%0A    public SessionEventDecoder sessionEventDecoder()%0A    %7B%0A        return sessionEventDecoder;%0A    %7D%0A%0A    public SessionHeaderDecoder sessionHeaderDecoder()%0A    %7B%0A        return sessionHeaderDecoder;%0A    %7D%0A%0A    public NewLeaderEventDecoder newLeaderEventDecoder()%0A    %7B%0A        return newLeaderEventDecoder;%0A    %7D
 %0A%7D%0A

Remove access to decoders.