<commit_before>/*
 * 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.qpid.proton4j.engine.impl;

import java.util.Objects;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.qpid.proton4j.amqp.Symbol;
import org.apache.qpid.proton4j.amqp.transport.AMQPHeader;
import org.apache.qpid.proton4j.amqp.transport.ErrorCondition;
import org.apache.qpid.proton4j.amqp.transport.Performative;
import org.apache.qpid.proton4j.buffer.ProtonBuffer;
import org.apache.qpid.proton4j.buffer.ProtonByteBufferAllocator;
import org.apache.qpid.proton4j.common.logging.ProtonLogger;
import org.apache.qpid.proton4j.common.logging.ProtonLoggerFactory;
import org.apache.qpid.proton4j.engine.ConnectionState;
import org.apache.qpid.proton4j.engine.Engine;
import org.apache.qpid.proton4j.engine.EnginePipeline;
import org.apache.qpid.proton4j.engine.EngineSaslDriver;
import org.apache.qpid.proton4j.engine.EngineState;
import org.apache.qpid.proton4j.engine.EventHandler;
import org.apache.qpid.proton4j.engine.exceptions.EngineFailedException;
import org.apache.qpid.proton4j.engine.exceptions.EngineNotStartedException;
import org.apache.qpid.proton4j.engine.exceptions.EngineNotWritableException;
import org.apache.qpid.proton4j.engine.exceptions.EngineShutdownException;
import org.apache.qpid.proton4j.engine.exceptions.EngineStartedException;
import org.apache.qpid.proton4j.engine.exceptions.EngineStateException;
import org.apache.qpid.proton4j.engine.exceptions.IdleTimeoutException;
import org.apache.qpid.proton4j.engine.exceptions.ProtonExceptionSupport;

/**
 * The default proton4j Engine implementation.
 */
public class ProtonEngine implements Engine {

    private static final ProtonLogger LOG = ProtonLoggerFactory.getLogger(ProtonEngine.class);

    private static final ProtonBuffer EMPTY_FRAME_BUFFER =
        ProtonByteBufferAllocator.DEFAULT.wrap(new byte[] {0x00, 0x00, 0x00, 0x08, 0x02, 0x00, 0x00, 0x00});

    private final ProtonEnginePipeline pipeline =  new ProtonEnginePipeline(this);
    private final ProtonEnginePipelineProxy pipelineProxy = new ProtonEnginePipelineProxy(pipeline);
    private final ProtonEngineConfiguration configuration = new ProtonEngineConfiguration(this);
    private final ProtonConnection connection = new ProtonConnection(this);

    private EngineSaslDriver saslDriver = new ProtonEngineNoOpSaslDriver();

    private boolean writable;
    private EngineState state = EngineState.IDLE;
    private Throwable failureCause;
    private int inputSequence;
    private int outputSequence;

    // Idle Timeout Check data
    private ScheduledFuture<?> nextIdleTimeoutCheck;
    private ScheduledExecutorService idleTimeoutExecutor;
    private int lastInputSequence;
    private int lastOutputSequence;
    private long localIdleDeadline = 0;
    private long remoteIdleDeadline = 0;

    // Engine event points
    private EventHandler<ProtonBuffer> outputHandler;
    private EventHandler<Throwable> engineErrorHandler = (error) -> {
        LOG.warn("Engine encounted error and will shutdown: ", error);
    };

    @Override
    public ProtonConnection getConnection() {
        return connection;
    }

    @Override
    public boolean isWritable() {
        return writable;
    }

    @Override
    public boolean isShutdown() {
        return state.ordinal() >= EngineState.SHUTDOWN.ordinal();
    }

    @Override
    public boolean isFailed() {
        return state == EngineState.FAILED;
    }

    @Override
    public Throwable failureCause() {
        return failureCause;
    }

    @Override
    public EngineState state() {
        return state;
    }

    @Override
    public ProtonConnection start() throws EngineStateException {
        checkShutdownOrFailed("Cannot start an Engine that has already been shutdown or has failed.");

        if (state == EngineState.IDLE) {
            state = EngineState.STARTING;
            try {
                pipeline.fireEngineStarting();
                state = EngineState.STARTED;
                writable = true;
            } catch (Throwable error) {
                throw engineFailed(error);
            }
        }

        return connection;
    }

    @Override
    public ProtonEngine shutdown() {
        if (state.ordinal() < EngineState.SHUTTING_DOWN.ordinal()) {
            state = EngineState.SHUTDOWN;
            writable = false;

            // TODO - We aren't currently checking connection state, do we want to close if open ?

            if (nextIdleTimeoutCheck != null) {
                LOG.trace("Cancelling scheduled Idle Timeout Check");
                nextIdleTimeoutCheck.cancel(false);
                nextIdleTimeoutCheck = null;
            }

            try {
                pipeline.fireEngineStateChanged();
            } catch (Throwable ignored) {}
        }

        return this;
    }

    @Override
    public long tick(long currentTime) throws IllegalStateException, EngineStateException {
        checkShutdownOrFailed("Cannot tick an Egnine that has been shutdown or failed.");

        if (connection.getState() != ConnectionState.ACTIVE) {
            throw new IllegalStateException("Cannot tick on a Connection that is not opened or an engine that has been shut down.");
        }

        if (idleTimeoutExecutor != null) {
            throw new IllegalStateException("Automatic ticking previously initiated.");
        }

        return performTick(currentTime);
    }

    @Override
    public void tickAuto(ScheduledExecutorService executor) throws IllegalStateException, EngineStateException {
        checkShutdownOrFailed("Cannot start auto tick on an Engine that has been shutdown or failed");

        Objects.requireNonNull(executor);

        if (isShutdown() || connection.getState() != ConnectionState.ACTIVE) {
            throw new IllegalStateException("Cannot tick on a Connection that is not opened or an engine that has been shut down.");
        }

        if (idleTimeoutExecutor != null) {
            throw new IllegalStateException("Automatic ticking previously initiated.");
        }

        // TODO - As an additional feature of this method we could allow for calling before connection is
        //        opened such that it starts ticking either on open local and also checks as a response to
        //        remote open which seems might be needed anyway, see notes in IdleTimeoutCheck class.

        // Immediate run of the idle timeout check logic will decide afterwards when / if we should
        // reschedule the idle timeout processing.
        LOG.trace("Auto Idle Timeout Check being initiated");
        idleTimeoutExecutor = executor;
        idleTimeoutExecutor.execute(new IdleTimeoutCheck());
    }

    @Override
    public ProtonEngine ingest(ProtonBuffer input) throws EngineStateException {
        checkShutdownOrFailed("Cannot ingest data into an Engine that has been shutdown or failed");

        if (!isWritable()) {
            throw new EngineNotWritableException("Engine is currently not accepting new input");
        }

        try {
            int startIndex = input.getReadIndex();
            pipeline.fireRead(input);
            if (input.getReadIndex() != startIndex) {
                inputSequence++;<commit_msg>Fix error check in tickAuto when checking connection open<commit_after>/*
 * 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.qpid.proton4j.engine.impl;

import java.util.Objects;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.qpid.proton4j.amqp.Symbol;
import org.apache.qpid.proton4j.amqp.transport.AMQPHeader;
import org.apache.qpid.proton4j.amqp.transport.ErrorCondition;
import org.apache.qpid.proton4j.amqp.transport.Performative;
import org.apache.qpid.proton4j.buffer.ProtonBuffer;
import org.apache.qpid.proton4j.buffer.ProtonByteBufferAllocator;
import org.apache.qpid.proton4j.common.logging.ProtonLogger;
import org.apache.qpid.proton4j.common.logging.ProtonLoggerFactory;
import org.apache.qpid.proton4j.engine.ConnectionState;
import org.apache.qpid.proton4j.engine.Engine;
import org.apache.qpid.proton4j.engine.EnginePipeline;
import org.apache.qpid.proton4j.engine.EngineSaslDriver;
import org.apache.qpid.proton4j.engine.EngineState;
import org.apache.qpid.proton4j.engine.EventHandler;
import org.apache.qpid.proton4j.engine.exceptions.EngineFailedException;
import org.apache.qpid.proton4j.engine.exceptions.EngineNotStartedException;
import org.apache.qpid.proton4j.engine.exceptions.EngineNotWritableException;
import org.apache.qpid.proton4j.engine.exceptions.EngineShutdownException;
import org.apache.qpid.proton4j.engine.exceptions.EngineStartedException;
import org.apache.qpid.proton4j.engine.exceptions.EngineStateException;
import org.apache.qpid.proton4j.engine.exceptions.IdleTimeoutException;
import org.apache.qpid.proton4j.engine.exceptions.ProtonExceptionSupport;

/**
 * The default proton4j Engine implementation.
 */
public class ProtonEngine implements Engine {

    private static final ProtonLogger LOG = ProtonLoggerFactory.getLogger(ProtonEngine.class);

    private static final ProtonBuffer EMPTY_FRAME_BUFFER =
        ProtonByteBufferAllocator.DEFAULT.wrap(new byte[] {0x00, 0x00, 0x00, 0x08, 0x02, 0x00, 0x00, 0x00});

    private final ProtonEnginePipeline pipeline =  new ProtonEnginePipeline(this);
    private final ProtonEnginePipelineProxy pipelineProxy = new ProtonEnginePipelineProxy(pipeline);
    private final ProtonEngineConfiguration configuration = new ProtonEngineConfiguration(this);
    private final ProtonConnection connection = new ProtonConnection(this);

    private EngineSaslDriver saslDriver = new ProtonEngineNoOpSaslDriver();

    private boolean writable;
    private EngineState state = EngineState.IDLE;
    private Throwable failureCause;
    private int inputSequence;
    private int outputSequence;

    // Idle Timeout Check data
    private ScheduledFuture<?> nextIdleTimeoutCheck;
    private ScheduledExecutorService idleTimeoutExecutor;
    private int lastInputSequence;
    private int lastOutputSequence;
    private long localIdleDeadline = 0;
    private long remoteIdleDeadline = 0;

    // Engine event points
    private EventHandler<ProtonBuffer> outputHandler;
    private EventHandler<Throwable> engineErrorHandler = (error) -> {
        LOG.warn("Engine encounted error and will shutdown: ", error);
    };

    @Override
    public ProtonConnection getConnection() {
        return connection;
    }

    @Override
    public boolean isWritable() {
        return writable;
    }

    @Override
    public boolean isShutdown() {
        return state.ordinal() >= EngineState.SHUTDOWN.ordinal();
    }

    @Override
    public boolean isFailed() {
        return state == EngineState.FAILED;
    }

    @Override
    public Throwable failureCause() {
        return failureCause;
    }

    @Override
    public EngineState state() {
        return state;
    }

    @Override
    public ProtonConnection start() throws EngineStateException {
        checkShutdownOrFailed("Cannot start an Engine that has already been shutdown or has failed.");

        if (state == EngineState.IDLE) {
            state = EngineState.STARTING;
            try {
                pipeline.fireEngineStarting();
                state = EngineState.STARTED;
                writable = true;
            } catch (Throwable error) {
                throw engineFailed(error);
            }
        }

        return connection;
    }

    @Override
    public ProtonEngine shutdown() {
        if (state.ordinal() < EngineState.SHUTTING_DOWN.ordinal()) {
            state = EngineState.SHUTDOWN;
            writable = false;

            // TODO - We aren't currently checking connection state, do we want to close if open ?

            if (nextIdleTimeoutCheck != null) {
                LOG.trace("Cancelling scheduled Idle Timeout Check");
                nextIdleTimeoutCheck.cancel(false);
                nextIdleTimeoutCheck = null;
            }

            try {
                pipeline.fireEngineStateChanged();
            } catch (Throwable ignored) {}
        }

        return this;
    }

    @Override
    public long tick(long currentTime) throws IllegalStateException, EngineStateException {
        checkShutdownOrFailed("Cannot tick an Egnine that has been shutdown or failed.");

        if (connection.getState() != ConnectionState.ACTIVE) {
            throw new IllegalStateException("Cannot tick on a Connection that is not opened or an engine that has been shut down.");
        }

        if (idleTimeoutExecutor != null) {
            throw new IllegalStateException("Automatic ticking previously initiated.");
        }

        return performTick(currentTime);
    }

    @Override
    public void tickAuto(ScheduledExecutorService executor) throws IllegalStateException, EngineStateException {
        checkShutdownOrFailed("Cannot start auto tick on an Engine that has been shutdown or failed");

        Objects.requireNonNull(executor);

        if (connection.getState() != ConnectionState.ACTIVE) {
            throw new IllegalStateException("Cannot tick on a Connection that is not opened.");
        }

        if (idleTimeoutExecutor != null) {
            throw new IllegalStateException("Automatic ticking previously initiated.");
        }

        // TODO - As an additional feature of this method we could allow for calling before connection is
        //        opened such that it starts ticking either on open local and also checks as a response to
        //        remote open which seems might be needed anyway, see notes in IdleTimeoutCheck class.

        // Immediate run of the idle timeout check logic will decide afterwards when / if we should
        // reschedule the idle timeout processing.
        LOG.trace("Auto Idle Timeout Check being initiated");
        idleTimeoutExecutor = executor;
        idleTimeoutExecutor.execute(new IdleTimeoutCheck());
    }

    @Override
    public ProtonEngine ingest(ProtonBuffer input) throws EngineStateException {
        checkShutdownOrFailed("Cannot ingest data into an Engine that has been shutdown or failed");

        if (!isWritable()) {
            throw new EngineNotWritableException("Engine is currently not accepting new input");
        }

        try {
            int startIndex = input.getReadIndex();
            pipeline.fireRead(input);
            if (input.getReadIndex() != startIndex) {
                inputSequence++;
{'fraction_non_alphanumeric': 0.07557171916414966, 'fraction_numerical': 0.006657215065031128, 'mean_word_length': 3.208560311284047, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1923467', 'n_tokens_mistral': 4293, 'n_tokens_neox': 4170, 'n_words': 1240}