/*
 * 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 org.apache.asterix.external.operators;

import java.util.concurrent.TimeUnit;

import org.apache.asterix.active.ActiveRuntimeId;
import org.apache.asterix.active.ActiveSourceOperatorNodePushable;
import org.apache.asterix.active.EntityId;
import org.apache.asterix.external.api.ITypedAdapterFactory;
import org.apache.asterix.external.dataset.adapter.FeedAdapter;
import org.apache.hyracks.api.comm.IFrame;
import org.apache.hyracks.api.comm.VSizeFrame;
import org.apache.hyracks.api.context.IHyracksTaskContext;
import org.apache.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import org.apache.hyracks.api.exceptions.ErrorCode;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.util.CleanupUtils;
import org.apache.hyracks.api.util.HyracksConstants;
import org.apache.hyracks.dataflow.common.io.MessagingFrameTupleAppender;
import org.apache.hyracks.dataflow.common.utils.TaskUtil;
import org.apache.logging.log4j.Level;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

/**
 * The runtime for @see{FeedIntakeOperationDescriptor}.
 * Provides the core functionality to set up the artifacts for ingestion of a feed.
 * The artifacts are lazily activated when a feed receives a subscription request.
 */
public class FeedIntakeOperatorNodePushable extends ActiveSourceOperatorNodePushable {
    private static final Logger LOGGER = LogManager.getLogger();
    private final FeedIntakeOperatorDescriptor opDesc;
    private final FeedAdapter adapter;
    private boolean poisoned = false;

    public FeedIntakeOperatorNodePushable(IHyracksTaskContext ctx, EntityId feedId, ITypedAdapterFactory adapterFactory,
            int partition, IRecordDescriptorProvider recordDescProvider,
            FeedIntakeOperatorDescriptor feedIntakeOperatorDescriptor) throws HyracksDataException {
        super(ctx, new ActiveRuntimeId(feedId, FeedIntakeOperatorNodePushable.class.getSimpleName(), partition));
        this.opDesc = feedIntakeOperatorDescriptor;
        this.recordDesc = recordDescProvider.getOutputRecordDescriptor(opDesc.getActivityId(), 0);
        adapter = (FeedAdapter) adapterFactory.createAdapter(ctx, runtimeId.getPartition());
    }

    @Override
    protected void start() throws HyracksDataException, InterruptedException {
        Throwable failure = null;
        Thread.currentThread().setName("Intake Thread");
        try {
            writer.open();
            synchronized (this) {
                if (poisoned) {
                    return;
                }
            }
            /*
             * Set null feed message. Feed pipeline carries with it a message with each frame
             * Initially, the message is set to a null message that can be changed by feed adapters.
             * One use case is adapters which consume data sources that allow restartability. Such adapters
             * can propagate progress information through the ingestion pipeline to storage nodes
             */
            IFrame message = new VSizeFrame(ctx);
            TaskUtil.put(HyracksConstants.KEY_MESSAGE, message, ctx);
            message.getBuffer().put(MessagingFrameTupleAppender.NULL_FEED_MESSAGE);
            message.getBuffer().flip();
            run();
        } catch (Throwable e) {
            failure = e;
            CleanupUtils.fail(writer, e);
            LOGGER.log(Level.WARN, "Failure during data ingestion", e);
        } finally {
            failure = CleanupUtils.close(adapter, failure);
            failure = CleanupUtils.close(writer, failure);
        }
        if (failure != null) {
            throw HyracksDataException.create(failure);
        }
    }

    private void run() throws HyracksDataException {
        // Start by getting the partition number from the manager
        LOGGER.info("Starting ingestion for partition:" + ctx.getTaskAttemptId().getTaskId().getPartition());
        try {
            doRun();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw HyracksDataException.create(e);
        } catch (Exception e) {
            LOGGER.log(Level.WARN, "Unhandled Exception", e);
            throw HyracksDataException.create(e);
        }
    }

    private void doRun() throws HyracksDataException, InterruptedException {
        while (true) {
            try {
                // Start the adapter
                adapter.start(ctx.getTaskAttemptId().getTaskId().getPartition(), writer);
                // Adapter has completed execution
                return;
            } catch (InterruptedException e) {
                throw e;
            } catch (Exception e) {
                LOGGER.log(Level.WARN, "Exception during feed ingestion ", e);
                throw HyracksDataException.create(e);
            }
        }
    }

    @Override
    protected void abort(long timeout, TimeUnit unit) throws HyracksDataException, InterruptedException {
        LOGGER.info(runtimeId + " aborting...");
        synchronized (this) {
            poisoned = true;
            try {
                if (!adapter.stop(unit.toMillis(timeout))) {
                    LOGGER.info(runtimeId + " failed to stop adapter. interrupting the thread...");
                    taskThread.interrupt();
                }
            } catch (HyracksDataException hde) {
                if (hde.matches(ErrorCode.TIMEOUT)) {
                    LOGGER.log(Level.WARN, runtimeId + " stop adapter timed out. interrupting the thread...", hde);
                    taskThread.interrupt();
                } else {
                    LOGGER.log(Level.WARN, "Failure during attempt to stop " + runtimeId, hde);
                    throw hde;
                }
            }
        }
    }

    @Override
    public String getStats() {
        if (adapter != null) {
            return "{\"adapter-stats\": " + adapter.getStats() + "}";
        } else {
            return "\"Runtime stats is not available.\"";
        }
    }
}
