/*
 * Copyright 2019 Mitsunori Komatsu (komamitsu)
 *
 * 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 org.komamitsu.fluency.fluentd.ingester.sender;

import org.komamitsu.fluency.ingester.sender.ErrorHandler;
import org.komamitsu.fluency.ingester.sender.Sender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public abstract class FluentdSender
        implements Closeable, Sender
{
    private static final Logger LOG = LoggerFactory.getLogger(FluentdSender.class);
    private final Config config;

    protected FluentdSender(Config config)
    {
        this.config = config;
    }

    protected FluentdSender()
    {
        this(new FluentdSender.Config());
    }

    public synchronized void send(ByteBuffer buffer)
            throws IOException
    {
        sendInternalWithRestoreBufferPositions(Arrays.asList(buffer), null);
    }

    public synchronized void send(List<ByteBuffer> buffers)
            throws IOException
    {
        sendInternalWithRestoreBufferPositions(buffers, null);
    }

    public void sendWithAck(List<ByteBuffer> buffers, String ackToken)
            throws IOException
    {
        sendInternalWithRestoreBufferPositions(buffers, ackToken);
    }

    private void sendInternalWithRestoreBufferPositions(List<ByteBuffer> buffers, String ackToken)
            throws IOException
    {
        List<Integer> positions = new ArrayList<>(buffers.size());
        for (ByteBuffer data : buffers) {
            positions.add(data.position());
        }

        try {
            sendInternal(buffers, ackToken);
        }
        catch (Exception e) {
            for (int i = 0; i < buffers.size(); i++) {
                buffers.get(i).position(positions.get(i));
            }

            ErrorHandler errorHandler = config.getErrorHandler();
            if (errorHandler != null) {
                try {
                    errorHandler.handle(e);
                }
                catch (Exception ex) {
                    LOG.warn("Failed to handle an error in the error handler {}", errorHandler, ex);
                }
            }

            if (e instanceof IOException) {
                throw (IOException) e;
            }
            else {
                throw new IOException(e);
            }
        }
    }

    public abstract boolean isAvailable();

    abstract protected void sendInternal(List<ByteBuffer> buffers, String ackToken)
            throws IOException;

    public static class Config
            extends Sender.Config
    {
    }
}
