<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.activemq.broker.openwire;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;

import org.apache.activemq.Service;
import org.apache.activemq.apollo.WindowLimiter;
import org.apache.activemq.apollo.broker.Broker;
import org.apache.activemq.apollo.broker.BrokerConnection;
import org.apache.activemq.apollo.broker.BrokerMessageDelivery;
import org.apache.activemq.apollo.broker.BrokerSubscription;
import org.apache.activemq.apollo.broker.Destination;
import org.apache.activemq.apollo.broker.MessageDelivery;
import org.apache.activemq.apollo.broker.ProtocolHandler;
import org.apache.activemq.apollo.broker.Router;
import org.apache.activemq.apollo.broker.VirtualHost;
import org.apache.activemq.broker.openwire.OpenWireMessageDelivery.PersistListener;
import org.apache.activemq.broker.store.Store.MessageRecord;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.BrokerId;
import org.apache.activemq.command.BrokerInfo;
import org.apache.activemq.command.Command;
import org.apache.activemq.command.ConnectionControl;
import org.apache.activemq.command.ConnectionError;
import org.apache.activemq.command.ConnectionId;
import org.apache.activemq.command.ConnectionInfo;
import org.apache.activemq.command.ConsumerControl;
import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.ControlCommand;
import org.apache.activemq.command.DestinationInfo;
import org.apache.activemq.command.ExceptionResponse;
import org.apache.activemq.command.FlushCommand;
import org.apache.activemq.command.KeepAliveInfo;
import org.apache.activemq.command.Message;
import org.apache.activemq.command.MessageAck;
import org.apache.activemq.command.MessageDispatch;
import org.apache.activemq.command.MessageDispatchNotification;
import org.apache.activemq.command.MessageId;
import org.apache.activemq.command.MessagePull;
import org.apache.activemq.command.ProducerAck;
import org.apache.activemq.command.ProducerId;
import org.apache.activemq.command.ProducerInfo;
import org.apache.activemq.command.RemoveInfo;
import org.apache.activemq.command.RemoveSubscriptionInfo;
import org.apache.activemq.command.Response;
import org.apache.activemq.command.SessionId;
import org.apache.activemq.command.SessionInfo;
import org.apache.activemq.command.ShutdownInfo;
import org.apache.activemq.command.TransactionInfo;
import org.apache.activemq.command.WireFormatInfo;
import org.apache.activemq.filter.BooleanExpression;
import org.apache.activemq.filter.FilterException;
import org.apache.activemq.filter.LogicExpression;
import org.apache.activemq.filter.NoLocalExpression;
import org.apache.activemq.flow.AbstractLimitedFlowResource;
import org.apache.activemq.flow.Flow;
import org.apache.activemq.flow.FlowController;
import org.apache.activemq.flow.IFlowController;
import org.apache.activemq.flow.IFlowLimiter;
import org.apache.activemq.flow.IFlowResource;
import org.apache.activemq.flow.IFlowSink;
import org.apache.activemq.flow.ISourceController;
import org.apache.activemq.flow.SizeLimiter;
import org.apache.activemq.flow.ISinkController.FlowControllable;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.protobuf.Buffer;
import org.apache.activemq.selector.SelectorParser;
import org.apache.activemq.state.CommandVisitor;
import org.apache.activemq.transport.WireFormatNegotiator;
import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.wireformat.WireFormat;

public class OpenwireProtocolHandler implements ProtocolHandler, PersistListener {

    protected final HashMap<ConnectionId, ClientContext> connections = new HashMap<ConnectionId, ClientContext>();
    protected final HashMap<SessionId, ClientContext> sessions = new HashMap<SessionId, ClientContext>();
    protected final HashMap<ProducerId, ProducerContext> producers = new HashMap<ProducerId, ProducerContext>();
    protected final HashMap<ConsumerId, ConsumerContext> consumers = new HashMap<ConsumerId, ConsumerContext>();

    protected BrokerConnection connection;
    private OpenWireFormat wireFormat;
    private OpenWireFormat storeWireFormat;
    private Router router;
    private VirtualHost host;
    private final CommandVisitor visitor;

    ArrayList<ActiveMQDestination> temporaryDestinations = new ArrayList<ActiveMQDestination>();

    public OpenwireProtocolHandler() {
        setStoreWireFormat(new OpenWireFormat());

        visitor = new CommandVisitor() {

            // /////////////////////////////////////////////////////////////////
            // Methods that keep track of the client state
            // /////////////////////////////////////////////////////////////////
            public Response processAddConnection(final ConnectionInfo info) throws Exception {
                if (!connections.containsKey(info.getConnectionId())) {

                    ClientContext connection = new AbstractClientContext<MessageDelivery>(info.getConnectionId().toString(), null) {
                        ConnectionInfo connectionInfo = info;

                        public void close() {
                            super.close();
                            connections.remove(connectionInfo.getConnectionId());
                        }
                    };
                    connections.put(info.getConnectionId(), connection);
                }
                return ack(info);
            }

            public Response processAddSession(final SessionInfo info) throws Exception {
                ClientContext connection = connections.get(info.getSessionId().getParentId());
                if (connection == null) {
                    throw new IllegalStateException(host.getHostName() + " Cannot add a session to a connection that had not been registered: " + info.getSessionId().getParentId());
                }

                if (!sessions.containsKey(info.getSessionId())) {

                    ClientContext session = new AbstractClientContext<MessageDelivery>(info.getSessionId().toString(), connection) {
                        SessionInfo sessioninfo = info;

                        public void close() {
                            super.close();
                            sessions.remove(sessioninfo.getSessionId());
                        }
                    };

                    sessions.put(info.getSessionId(), session);
                }

                return ack(info);
            }

            public Response processAddProducer(ProducerInfo info) throws Exception {
                ClientContext session = sessions.get(info.getProducerId().getParentId());
                if (session == null) {
                    throw new IllegalStateException(host.getHostName() + " Cannot add a producer to a session that had not been registered: " + info.getProducerId().getParentId());
                }
                if (!producers.containsKey(info.getProducerId())) {
                    ProducerContext producer = new ProducerContext(info, session);
                }
                return ack(info);
            }
<commit_msg>Adding the default session.. Makes the JmsRedeliveredTest test happier.<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.activemq.broker.openwire;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;

import org.apache.activemq.apollo.WindowLimiter;
import org.apache.activemq.apollo.broker.Broker;
import org.apache.activemq.apollo.broker.BrokerConnection;
import org.apache.activemq.apollo.broker.BrokerMessageDelivery;
import org.apache.activemq.apollo.broker.BrokerSubscription;
import org.apache.activemq.apollo.broker.Destination;
import org.apache.activemq.apollo.broker.MessageDelivery;
import org.apache.activemq.apollo.broker.ProtocolHandler;
import org.apache.activemq.apollo.broker.Router;
import org.apache.activemq.apollo.broker.VirtualHost;
import org.apache.activemq.broker.openwire.OpenWireMessageDelivery.PersistListener;
import org.apache.activemq.broker.store.Store.MessageRecord;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.BrokerId;
import org.apache.activemq.command.BrokerInfo;
import org.apache.activemq.command.Command;
import org.apache.activemq.command.ConnectionControl;
import org.apache.activemq.command.ConnectionError;
import org.apache.activemq.command.ConnectionId;
import org.apache.activemq.command.ConnectionInfo;
import org.apache.activemq.command.ConsumerControl;
import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.ControlCommand;
import org.apache.activemq.command.DestinationInfo;
import org.apache.activemq.command.ExceptionResponse;
import org.apache.activemq.command.FlushCommand;
import org.apache.activemq.command.KeepAliveInfo;
import org.apache.activemq.command.Message;
import org.apache.activemq.command.MessageAck;
import org.apache.activemq.command.MessageDispatch;
import org.apache.activemq.command.MessageDispatchNotification;
import org.apache.activemq.command.MessageId;
import org.apache.activemq.command.MessagePull;
import org.apache.activemq.command.ProducerAck;
import org.apache.activemq.command.ProducerId;
import org.apache.activemq.command.ProducerInfo;
import org.apache.activemq.command.RemoveInfo;
import org.apache.activemq.command.RemoveSubscriptionInfo;
import org.apache.activemq.command.Response;
import org.apache.activemq.command.SessionId;
import org.apache.activemq.command.SessionInfo;
import org.apache.activemq.command.ShutdownInfo;
import org.apache.activemq.command.TransactionInfo;
import org.apache.activemq.command.WireFormatInfo;
import org.apache.activemq.filter.BooleanExpression;
import org.apache.activemq.filter.FilterException;
import org.apache.activemq.filter.LogicExpression;
import org.apache.activemq.filter.NoLocalExpression;
import org.apache.activemq.flow.Flow;
import org.apache.activemq.flow.FlowController;
import org.apache.activemq.flow.IFlowController;
import org.apache.activemq.flow.IFlowLimiter;
import org.apache.activemq.flow.IFlowResource;
import org.apache.activemq.flow.IFlowSink;
import org.apache.activemq.flow.ISourceController;
import org.apache.activemq.flow.SizeLimiter;
import org.apache.activemq.flow.ISinkController.FlowControllable;
import org.apache.activemq.openwire.OpenWireFormat;
import org.apache.activemq.protobuf.Buffer;
import org.apache.activemq.selector.SelectorParser;
import org.apache.activemq.state.CommandVisitor;
import org.apache.activemq.transport.WireFormatNegotiator;
import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.wireformat.WireFormat;

public class OpenwireProtocolHandler implements ProtocolHandler, PersistListener {

    protected final HashMap<ConnectionId, ClientContext> connections = new HashMap<ConnectionId, ClientContext>();
    protected final HashMap<SessionId, ClientContext> sessions = new HashMap<SessionId, ClientContext>();
    protected final HashMap<ProducerId, ProducerContext> producers = new HashMap<ProducerId, ProducerContext>();
    protected final HashMap<ConsumerId, ConsumerContext> consumers = new HashMap<ConsumerId, ConsumerContext>();

    protected BrokerConnection connection;
    private OpenWireFormat wireFormat;
    private OpenWireFormat storeWireFormat;
    private Router router;
    private VirtualHost host;
    private final CommandVisitor visitor;

    ArrayList<ActiveMQDestination> temporaryDestinations = new ArrayList<ActiveMQDestination>();

    public OpenwireProtocolHandler() {
        setStoreWireFormat(new OpenWireFormat());

        visitor = new CommandVisitor() {

            // /////////////////////////////////////////////////////////////////
            // Methods that keep track of the client state
            // /////////////////////////////////////////////////////////////////
            public Response processAddConnection(final ConnectionInfo info) throws Exception {
                if (!connections.containsKey(info.getConnectionId())) {

                    ClientContext connection = new AbstractClientContext<MessageDelivery>(info.getConnectionId().toString(), null) {
                        ConnectionInfo connectionInfo = info;

                        public void close() {
                            super.close();
                            connections.remove(connectionInfo.getConnectionId());
                        }
                    };
                    connections.put(info.getConnectionId(), connection);
                    
                    // Connections have an implicitly created "default" session identified by session id = -1
                    SessionId sessionId = new SessionId(info.getConnectionId(), -1);
                    addSession(sessionId, connection);
                }
                return ack(info);
            }

            public Response processAddSession(final SessionInfo info) throws Exception {
                final SessionId sessionId = info.getSessionId();
                ClientContext connection = connections.get(sessionId.getParentId());
                if (connection == null) {
                    throw new IllegalStateException(host.getHostName() + " Cannot add a session to a connection that had not been registered: " + sessionId.getParentId());
                }

                if (!sessions.containsKey(sessionId)) {
                    addSession(sessionId, connection);
                }

                return ack(info);
            }

            private void addSession(final SessionId sessionId, ClientContext connection) {
                ClientContext session = new AbstractClientContext<MessageDelivery>(sessionId.toString(), connection) {
                    public void close() {
                        super.close();
                        sessions.remove(sessionId);
                    }
                };
                sessions.put(sessionId, session);
            }

            public Response processAddProducer(ProducerInfo info) throws Exception {
                ClientContext session = sessions.get(info.getProducerId().getParentId());
                if (session == null) {
                    throw new IllegalStateException(host.getHostName() + " Cannot add a producer to a session that had not been registered: " + info.getProducerId().getParentId());
                }
                if (!producers.containsKey(info.getProducerId())) {
                    ProducerContext producer = new ProducerContext(info, session);
                }
                return ack(info);
            }

{'fraction_non_alphanumeric': 0.10156728296804213, 'fraction_numerical': 0.0006122199093914534, 'mean_word_length': 3.2100515463917527, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, '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': '4632517', 'n_tokens_mistral': 4132, 'n_tokens_neox': 4011, 'n_words': 925}