// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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 streamer;

import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.LogManager;

/**
 * Link to transfer data in bounds of single thread (synchronized transfer).
 * Must not be used to send data to elements served in different threads.
 */
public class SyncLink implements Link {
    protected Logger logger = LogManager.getLogger(getClass());

    /**
     * When null packet is pulled from source element, then make slight delay to
     * avoid consuming of 100% of CPU in main loop in cases when link is pauses or
     * source element cannot produce data right now.
     */
    public static final long STANDARD_DELAY_FOR_EMPTY_PACKET = 10; // Milliseconds

    /**
     * Delay for null packets in poll method when blocking is requested, in
     * milliseconds.
     */
    protected long delay = STANDARD_DELAY_FOR_EMPTY_PACKET;

    /**
     * Set to true to print debugging messages.
     */
    protected boolean verbose = System.getProperty("streamer.Link.debug", "false").equals("true");;

    /**
     * ID of this link.
     */
    protected String id = null;

    /**
     * Buffer with data to hold because link is paused, on hold, or data is pushed
     * back from output element.
     */
    protected ByteBuffer cacheBuffer = null;

    /**
     * Size of expected packet. Data must be hold in link until full packet will
     * be read.
     */
    protected int expectedPacketSize = 0;

    /**
     * Number of packets and packet header transferred to element.
     */
    protected int packetNumber = 0;

    /**
     * Element to pull data from, when in pull mode.
     */
    protected Element source = null;

    /**
     * Element to send data to in both pull and push modes.
     */
    protected Element sink = null;

    /**
     * When in loop, indicates that loop must be stopped.
     *
     * @see run()
     */
    private boolean shutdown = false;

    /**
     * Indicates that event STREAM_START is passed over this link, so main loop
     * can be started to pull data from source element.
     */
    protected boolean started = false;

    /**
     * Set to true to hold all data in link until it will be set to false again.
     */
    protected boolean paused = false;

    /**
     * Used by pull() method to hold all data in this link to avoid recursion when
     * source element is asked to push new data to it outputs.
     */
    protected boolean hold = false;

    /**
     * Operate in pull mode instead of default push mode. In pull mode, link will
     * ask it source element for new data.
     */
    protected boolean pullMode = false;

    public SyncLink() {
    }

    public SyncLink(String id) {
        this.id = id;
    }

    @Override
    public void pushBack(ByteBuffer buf) {
        if (verbose)
            logger.debug("[" + this + "] INFO: Buffer pushed back: " + buf + ".");

        if (cacheBuffer != null) {
            ByteBuffer tmp = cacheBuffer.join(buf);
            cacheBuffer.unref();
            cacheBuffer = tmp;
        } else {
            cacheBuffer = buf;
            cacheBuffer.ref();
        }

        resetCursor();
    }

    private void resetCursor() {
        // Reset cursor
        cacheBuffer.cursor = 0;
    }

    @Override
    public void pushBack(ByteBuffer buf, int lengthOfFullPacket) {
        pushBack(buf);
        expectedPacketSize = lengthOfFullPacket;
    }

    @Override
    public String toString() {
        return "SyncLink(" + ((id != null) ? id + ", " : "") + source + ":" + sink + ")";
    }

    /**
     * Push data to sink. Call with null to push cached data.
     */
    @Override
    public void sendData(ByteBuffer buf) {
        if (!hold && pullMode)
            throw new RuntimeException("[" + this + "] ERROR: link is not in push mode.");

        if (verbose)
            logger.debug("[" + this + "] INFO: Incoming buffer: " + buf + ".");

        if (buf == null && cacheBuffer == null)
            return;

        if (cacheBuffer != null && buf != null) {
            // Join old data with fresh data
            buf = cacheBuffer.join(buf);
            cacheBuffer.unref();
            cacheBuffer = buf;
        }

        // Store buffer in cache field to simplify following loop
        if (buf != null)
            cacheBuffer = buf;

        // When data pushed back and length of data is less than length of full
        // packet, then feed data to sink element immediately
        while (cacheBuffer != null) {
            if (paused || hold) {
                if (verbose)
                    logger.debug("[" + this + "] INFO: Transfer is paused. Data in cache buffer: " + cacheBuffer + ".");

                // Wait until rest of packet will be read
                return;
            }

            if (expectedPacketSize > 0 && cacheBuffer.length < expectedPacketSize) {
                if (verbose)
                    logger.debug("[" + this + "] INFO: Transfer is suspended because available data is less than expected packet size. Expected packet size: "
                            + expectedPacketSize + ", data in cache buffer: " + cacheBuffer + ".");

                // Wait until rest of packet will be read
                return;
            }

            // Full packet or packet header is read, feed it to element
            buf = cacheBuffer;
            cacheBuffer = null;
            expectedPacketSize = 0;
            packetNumber++;

            if (sink == null)
                throw new NullPointerException("[" + this + "] ERROR: Cannot send data to sink: sink is null. Data: " + buf + ".");

            sink.handleData(buf, this);
            // cacheBuffer and expectedPacketSize can be changed at this time
        }

    }

    @SuppressWarnings("incomplete-switch")
    @Override
    public void sendEvent(Event event, Direction direction) {

        if (verbose)
            logger.debug("[" + this + "] INFO: Event " + event + " is received.");

        // Shutdown main loop (if any) when STREAM_CLOSE event is received.
        switch (event) {
        case STREAM_START: {
            if (!started)
                started = true;
            else
                // Event already sent trough this link
                return;
            break;
        }
        case STREAM_CLOSE: {
            if (!shutdown)
                shutdown = true;
            else
                // Event already sent trough this link
                return;
            break;
        }
        case LINK_SWITCH_TO_PULL_MODE: {
            setPullMode();
            break;
        }

        }

        switch (direction) {
        case IN:
            source.handleEvent(event, direction);
            break;
        case OUT:
            sink.handleEvent(event, direction);
            break;
        }
    }

    @Override
    public ByteBuffer pull(boolean block) {
        if (!pullMode)
            throw new RuntimeException("[" + this + "] ERROR: This link is not in pull mode.");

        if (hold)
            throw new RuntimeException("[" + this + "] ERROR: This link is already on hold, waiting for data to be pulled in. Circular reference?");

        if (paused) {
            if (verbose)
                logger.debug("[" + this + "] INFO: Cannot pull, link is paused.");

            // Make slight delay in such case, to avoid consuming 100% of CPU
            if (block) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    logger.info("[ignored] interrupted during pull", e);
                }
            }

            return null;
        }

        // If data in cache can be sent immediately,
        // then return it instead of asking for more data from source
        if (cacheBuffer != null && (expectedPacketSize == 0 || (expectedPacketSize > 0 && cacheBuffer.length >= expectedPacketSize))) {
            if (verbose)
                logger.debug("[" + this + "] INFO: Data pulled from cache buffer: " + cacheBuffer + ".");

            ByteBuffer tmp = cacheBuffer;
            cacheBuffer = null;
            return tmp;
        }

        // Pause this link, so incoming data will not be sent to sink
        // immediately, then ask source element for more data
        try {
            hold = true;
            source.poll(block);
        } finally {
            hold = false;
        }

        // Can return something only when data was stored in buffer
        if (cacheBuffer != null && (expectedPacketSize == 0 || (expectedPacketSize > 0 && cacheBuffer.length >= expectedPacketSize))) {
            if (verbose)
                logger.debug("[" + this + "] INFO: Data pulled from source: " + cacheBuffer + ".");

            ByteBuffer tmp = cacheBuffer;
            cacheBuffer = null;
            return tmp;
        } else {
            return null;
        }
    }

    @Override
    public Element setSink(Element sink) {
        if (sink != null && this.sink != null)
            throw new RuntimeException("[" + this + "] ERROR: This link sink element is already set. Link: " + this + ", new sink: " + sink + ", existing sink: "
                    + this.sink + ".");

        if (sink == null && cacheBuffer != null)
            throw new RuntimeException("[" + this + "] ERROR: Cannot drop link: cache is not empty. Link: " + this + ", cache: " + cacheBuffer);

        this.sink = sink;

        return sink;
    }

    @Override
    public Element setSource(Element source) {
        if (this.source != null && source != null)
            throw new RuntimeException("[" + this + "] ERROR: This link source element is already set. Link: " + this + ", new source: " + source
                    + ", existing source: " + this.source + ".");

        this.source = source;
        return source;
    }

    @Override
    public Element getSource() {
        return source;
    }

    @Override
    public Element getSink() {
        return sink;
    }

    @Override
    public void pause() {
        if (paused)
            throw new RuntimeException("[" + this + "] ERROR: Link is already paused.");

        paused = true;

    }

    @Override
    public void resume() {
        paused = false;
    }

    /**
     * Run pull loop to actively pull data from source and push it to sink. It
     * must be only one pull loop per thread.
     *
     * Pull loop will start after event STREAM_START. This link and source element
     * incoming links will be switched to pull mode before pull loop will be
     * started using event LINK_SWITCH_TO_PULL_MODE.
     */
    @Override
    public void run() {
        // Wait until even STREAM_START will arrive
        while (!started) {
            delay();
        }

        sendEvent(Event.LINK_SWITCH_TO_PULL_MODE, Direction.IN);

        if (verbose)
            logger.debug("[" + this + "] INFO: Starting pull loop.");

        // Pull source in loop
        while (!shutdown) {
            // Pull data from source element and send it to sink element
            ByteBuffer data = pull(false);
            if (data != null)
                sink.handleData(data, this);

            if (!shutdown && data == null) {
                // Make slight delay to avoid consuming of 100% of CPU
                delay();
            }
        }

        if (verbose)
            logger.debug("[" + this + "] INFO: Pull loop finished.");

    }

    protected void delay() {
        try {
            Thread.sleep(delay);
        } catch (InterruptedException e) {
            throw new RuntimeException("[" + this + "] ERROR: Interrupted in main loop.", e);
        }
    }

    @Override
    public void setPullMode() {
        if (verbose)
            logger.debug("[" + this + "] INFO: Switching to PULL mode.");

        pullMode = true;
    }

    @Override
    public void drop() {
        if (pullMode)
            throw new RuntimeException("[" + this + "] ERROR: Cannot drop link in pull mode.");

        if (cacheBuffer != null)
            throw new RuntimeException("[" + this + "] ERROR: Cannot drop link when cache contains data: " + cacheBuffer + ".");

        source.dropLink(this);
        sink.dropLink(this);
    }
}