<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 com.ibm.mqlight.api.impl.engine;

import io.netty.buffer.ByteBuf;

import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.EnumSet;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.qpid.proton.Proton;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.messaging.Modified;
import org.apache.qpid.proton.amqp.messaging.Rejected;
import org.apache.qpid.proton.amqp.messaging.Released;
import org.apache.qpid.proton.amqp.messaging.Source;
import org.apache.qpid.proton.amqp.messaging.Target;
import org.apache.qpid.proton.amqp.messaging.TerminusExpiryPolicy;
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode;
import org.apache.qpid.proton.amqp.transport.SenderSettleMode;
import org.apache.qpid.proton.engine.Collector;
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.EndpointState;
import org.apache.qpid.proton.engine.Event;
import org.apache.qpid.proton.engine.Handler;
import org.apache.qpid.proton.engine.Link;
import org.apache.qpid.proton.engine.Receiver;
import org.apache.qpid.proton.engine.Sasl;
import org.apache.qpid.proton.engine.Sender;
import org.apache.qpid.proton.engine.Session;
import org.apache.qpid.proton.engine.Transport;

import com.ibm.mqlight.api.ClientException;
import com.ibm.mqlight.api.NetworkException;
import com.ibm.mqlight.api.NotPermittedException;
import com.ibm.mqlight.api.Promise;
import com.ibm.mqlight.api.QOS;
import com.ibm.mqlight.api.ReplacedException;
import com.ibm.mqlight.api.StateException;
import com.ibm.mqlight.api.SubscribedException;
import com.ibm.mqlight.api.UnsubscribedException;
import com.ibm.mqlight.api.impl.ComponentImpl;
import com.ibm.mqlight.api.impl.Message;
import com.ibm.mqlight.api.impl.SubscriptionTopic;
import com.ibm.mqlight.api.impl.network.ConnectResponse;
import com.ibm.mqlight.api.impl.network.ConnectionError;
import com.ibm.mqlight.api.impl.network.DataRead;
import com.ibm.mqlight.api.impl.network.DisconnectResponse;
import com.ibm.mqlight.api.impl.network.NetworkClosePromiseImpl;
import com.ibm.mqlight.api.impl.network.NetworkConnectPromiseImpl;
import com.ibm.mqlight.api.impl.network.NetworkListenerImpl;
import com.ibm.mqlight.api.impl.network.NetworkWritePromiseImpl;
import com.ibm.mqlight.api.impl.network.WriteResponse;
import com.ibm.mqlight.api.impl.timer.PopResponse;
import com.ibm.mqlight.api.impl.timer.TimerPromiseImpl;
import com.ibm.mqlight.api.logging.FFDCProbeId;
import com.ibm.mqlight.api.logging.Logger;
import com.ibm.mqlight.api.logging.LoggerFactory;
import com.ibm.mqlight.api.network.NetworkChannel;
import com.ibm.mqlight.api.network.NetworkService;
import com.ibm.mqlight.api.timer.TimerService;

public class Engine extends ComponentImpl implements Handler {

    private static final Logger logger = LoggerFactory.getLogger(Engine.class);

    private final NetworkService network;
    private final TimerService timer;

    public Engine(NetworkService network, TimerService timer) {
        final String methodName = "<init>";
        logger.entry(this, methodName, network, timer);

        if (network == null) {
          final IllegalArgumentException exception = new IllegalArgumentException("NetworkService argument cannot be null");
          logger.throwing(this, methodName, exception);
          throw exception;
        }
        if (timer == null) {
          final IllegalArgumentException exception = new IllegalArgumentException("TimerService argument cannot be null");
          logger.throwing(this, methodName, exception);
          throw exception;
        }
        this.network = network;
        this.timer = timer;

        logger.exit(this, methodName);
    }

    @Override
    protected void onReceive(Message message) {
        final String methodName = "onReceive";
        logger.entry(this, methodName, message);

        if (message instanceof OpenRequest) {
            OpenRequest or = (OpenRequest)message;
            NetworkListenerImpl listener = new NetworkListenerImpl(this);
            Promise<NetworkChannel> promise = new NetworkConnectPromiseImpl(this, or);
            network.connect(or.endpoint, listener, promise);
        }
        else if (message instanceof ConnectResponse) {
            // Message from network telling us that a connect request has completed...
            ConnectResponse cr = (ConnectResponse)message;
            OpenRequest or = (OpenRequest)cr.context;
            if (cr.exception != null) {
                or.getSender().tell(new OpenResponse(or, cr.exception), this);
            } else {
                Connection protonConnection = Proton.connection();
                Transport transport = Proton.transport();
                transport.setIdleTimeout(or.endpoint.getIdleTimeout());
                transport.bind(protonConnection);
                Collector collector = Proton.collector();
                protonConnection.setContainer(or.clientId);
                protonConnection.setHostname(or.endpoint.getHost());
                protonConnection.open();
                Sasl sasl = transport.sasl();
                sasl.client();
                if (or.endpoint.getUser() == null) {
                    sasl.setMechanisms(new String[]{"ANONYMOUS"});
                } else {
                    sasl.plain(or.endpoint.getUser(), or.endpoint.getPassword());
                }
                Session session = protonConnection.session();
                session.open();
                protonConnection.collect(collector);

                EngineConnection engineConnection = new EngineConnection(protonConnection, session, or.getSender(), transport, collector, cr.channel);
                engineConnection.openRequest = or;
                protonConnection.setContext(engineConnection);
                cr.channel.setContext(engineConnection);

                // Write any data from Proton to the network.
                writeToNetwork(engineConnection);
            }
        } else if (message instanceof CloseRequest) {
            CloseRequest cr = (CloseRequest)message;
            Connection protonConnection = cr.connection.connection;
            EngineConnection engineConnection = (EngineConnection)protonConnection.getContext();
            if (engineConnection.timerPromise != null) {
                TimerPromiseImpl tmp = engineConnection.timerPromise;
                engineConnection.timerPromise = null;
                timer.cancel(tmp);
            }
            protonConnection.close();
            engineConnection.closeRequest = cr;
            writeToNetwork(engineConnection);
        } else if (message instanceof SendRequest) {
            SendRequest sr = (SendRequest)message;
            EngineConnection engineConnection = sr.connection;

            // Look to see if there is already a suitable sending link, and open one if there is not...
            Link link = sr.connection.connection.linkHead(EnumSet.of(EndpointState.ACTIVE),
                                                          EnumSet.of(EndpointState.ACTIVE, EndpointState.UNINITIALIZED));
            Sender linkSender;
            boolean linkOpened = false;
            while(true) {
                if (link == null) {
                    linkSender = sr.connection.session.sender(sr.topic);
                    Source source = new Source();
                    Target target = new Target();
                    source.setAddress(sr.topic);
                    target.setAddress(sr.topic);
                    linkSender.setSource(source);
                    linkSender.setTarget(target);
                    linkSender.open();
                    linkOpened = true;
                    break;
                }
                if ((link instanceof Sender) && sr.topic.equals(link.getName())) {
                    sr.topic.equals(link.getName());
                    linkSender = (Sender)link;<commit_msg>Removed dead call to topic.equals<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 com.ibm.mqlight.api.impl.engine;

import io.netty.buffer.ByteBuf;

import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.EnumSet;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.qpid.proton.Proton;
import org.apache.qpid.proton.amqp.Symbol;
import org.apache.qpid.proton.amqp.UnsignedInteger;
import org.apache.qpid.proton.amqp.messaging.Modified;
import org.apache.qpid.proton.amqp.messaging.Rejected;
import org.apache.qpid.proton.amqp.messaging.Released;
import org.apache.qpid.proton.amqp.messaging.Source;
import org.apache.qpid.proton.amqp.messaging.Target;
import org.apache.qpid.proton.amqp.messaging.TerminusExpiryPolicy;
import org.apache.qpid.proton.amqp.transport.ErrorCondition;
import org.apache.qpid.proton.amqp.transport.ReceiverSettleMode;
import org.apache.qpid.proton.amqp.transport.SenderSettleMode;
import org.apache.qpid.proton.engine.Collector;
import org.apache.qpid.proton.engine.Connection;
import org.apache.qpid.proton.engine.Delivery;
import org.apache.qpid.proton.engine.EndpointState;
import org.apache.qpid.proton.engine.Event;
import org.apache.qpid.proton.engine.Handler;
import org.apache.qpid.proton.engine.Link;
import org.apache.qpid.proton.engine.Receiver;
import org.apache.qpid.proton.engine.Sasl;
import org.apache.qpid.proton.engine.Sender;
import org.apache.qpid.proton.engine.Session;
import org.apache.qpid.proton.engine.Transport;

import com.ibm.mqlight.api.ClientException;
import com.ibm.mqlight.api.NetworkException;
import com.ibm.mqlight.api.NotPermittedException;
import com.ibm.mqlight.api.Promise;
import com.ibm.mqlight.api.QOS;
import com.ibm.mqlight.api.ReplacedException;
import com.ibm.mqlight.api.StateException;
import com.ibm.mqlight.api.SubscribedException;
import com.ibm.mqlight.api.UnsubscribedException;
import com.ibm.mqlight.api.impl.ComponentImpl;
import com.ibm.mqlight.api.impl.Message;
import com.ibm.mqlight.api.impl.SubscriptionTopic;
import com.ibm.mqlight.api.impl.network.ConnectResponse;
import com.ibm.mqlight.api.impl.network.ConnectionError;
import com.ibm.mqlight.api.impl.network.DataRead;
import com.ibm.mqlight.api.impl.network.DisconnectResponse;
import com.ibm.mqlight.api.impl.network.NetworkClosePromiseImpl;
import com.ibm.mqlight.api.impl.network.NetworkConnectPromiseImpl;
import com.ibm.mqlight.api.impl.network.NetworkListenerImpl;
import com.ibm.mqlight.api.impl.network.NetworkWritePromiseImpl;
import com.ibm.mqlight.api.impl.network.WriteResponse;
import com.ibm.mqlight.api.impl.timer.PopResponse;
import com.ibm.mqlight.api.impl.timer.TimerPromiseImpl;
import com.ibm.mqlight.api.logging.FFDCProbeId;
import com.ibm.mqlight.api.logging.Logger;
import com.ibm.mqlight.api.logging.LoggerFactory;
import com.ibm.mqlight.api.network.NetworkChannel;
import com.ibm.mqlight.api.network.NetworkService;
import com.ibm.mqlight.api.timer.TimerService;

public class Engine extends ComponentImpl implements Handler {

    private static final Logger logger = LoggerFactory.getLogger(Engine.class);

    private final NetworkService network;
    private final TimerService timer;

    public Engine(NetworkService network, TimerService timer) {
        final String methodName = "<init>";
        logger.entry(this, methodName, network, timer);

        if (network == null) {
          final IllegalArgumentException exception = new IllegalArgumentException("NetworkService argument cannot be null");
          logger.throwing(this, methodName, exception);
          throw exception;
        }
        if (timer == null) {
          final IllegalArgumentException exception = new IllegalArgumentException("TimerService argument cannot be null");
          logger.throwing(this, methodName, exception);
          throw exception;
        }
        this.network = network;
        this.timer = timer;

        logger.exit(this, methodName);
    }

    @Override
    protected void onReceive(Message message) {
        final String methodName = "onReceive";
        logger.entry(this, methodName, message);

        if (message instanceof OpenRequest) {
            OpenRequest or = (OpenRequest)message;
            NetworkListenerImpl listener = new NetworkListenerImpl(this);
            Promise<NetworkChannel> promise = new NetworkConnectPromiseImpl(this, or);
            network.connect(or.endpoint, listener, promise);
        }
        else if (message instanceof ConnectResponse) {
            // Message from network telling us that a connect request has completed...
            ConnectResponse cr = (ConnectResponse)message;
            OpenRequest or = (OpenRequest)cr.context;
            if (cr.exception != null) {
                or.getSender().tell(new OpenResponse(or, cr.exception), this);
            } else {
                Connection protonConnection = Proton.connection();
                Transport transport = Proton.transport();
                transport.setIdleTimeout(or.endpoint.getIdleTimeout());
                transport.bind(protonConnection);
                Collector collector = Proton.collector();
                protonConnection.setContainer(or.clientId);
                protonConnection.setHostname(or.endpoint.getHost());
                protonConnection.open();
                Sasl sasl = transport.sasl();
                sasl.client();
                if (or.endpoint.getUser() == null) {
                    sasl.setMechanisms(new String[]{"ANONYMOUS"});
                } else {
                    sasl.plain(or.endpoint.getUser(), or.endpoint.getPassword());
                }
                Session session = protonConnection.session();
                session.open();
                protonConnection.collect(collector);

                EngineConnection engineConnection = new EngineConnection(protonConnection, session, or.getSender(), transport, collector, cr.channel);
                engineConnection.openRequest = or;
                protonConnection.setContext(engineConnection);
                cr.channel.setContext(engineConnection);

                // Write any data from Proton to the network.
                writeToNetwork(engineConnection);
            }
        } else if (message instanceof CloseRequest) {
            CloseRequest cr = (CloseRequest)message;
            Connection protonConnection = cr.connection.connection;
            EngineConnection engineConnection = (EngineConnection)protonConnection.getContext();
            if (engineConnection.timerPromise != null) {
                TimerPromiseImpl tmp = engineConnection.timerPromise;
                engineConnection.timerPromise = null;
                timer.cancel(tmp);
            }
            protonConnection.close();
            engineConnection.closeRequest = cr;
            writeToNetwork(engineConnection);
        } else if (message instanceof SendRequest) {
            SendRequest sr = (SendRequest)message;
            EngineConnection engineConnection = sr.connection;

            // Look to see if there is already a suitable sending link, and open one if there is not...
            Link link = sr.connection.connection.linkHead(EnumSet.of(EndpointState.ACTIVE),
                                                          EnumSet.of(EndpointState.ACTIVE, EndpointState.UNINITIALIZED));
            Sender linkSender;
            boolean linkOpened = false;
            while(true) {
                if (link == null) {
                    linkSender = sr.connection.session.sender(sr.topic);
                    Source source = new Source();
                    Target target = new Target();
                    source.setAddress(sr.topic);
                    target.setAddress(sr.topic);
                    linkSender.setSource(source);
                    linkSender.setTarget(target);
                    linkSender.open();
                    linkOpened = true;
                    break;
                }
                if ((link instanceof Sender) && sr.topic.equals(link.getName())) {
                    linkSender = (Sender)link;
{'fraction_non_alphanumeric': 0.0948969959464712, 'fraction_numerical': 0.00044422233327780556, 'mean_word_length': 2.8060016906170753, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2560868', 'n_tokens_mistral': 4938, 'n_tokens_neox': 4720, 'n_words': 1044}