<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.region;

import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

import javax.jms.InvalidClientIDException;
import javax.jms.JMSException;

import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.Connection;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.ConsumerBrokerExchange;
import org.apache.activemq.broker.ProducerBrokerExchange;
import org.apache.activemq.broker.region.policy.DeadLetterStrategy;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.BrokerId;
import org.apache.activemq.command.BrokerInfo;
import org.apache.activemq.command.ConnectionId;
import org.apache.activemq.command.ConnectionInfo;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.DestinationInfo;
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.MessagePull;
import org.apache.activemq.command.ProducerInfo;
import org.apache.activemq.command.RemoveSubscriptionInfo;
import org.apache.activemq.command.Response;
import org.apache.activemq.command.SessionInfo;
import org.apache.activemq.command.TransactionId;
import org.apache.activemq.kaha.Store;
import org.apache.activemq.memory.UsageManager;
import org.apache.activemq.store.PersistenceAdapter;
import org.apache.activemq.store.memory.MemoryPersistenceAdapter;
import org.apache.activemq.thread.TaskRunnerFactory;
import org.apache.activemq.util.BrokerSupport;
import org.apache.activemq.util.IdGenerator;
import org.apache.activemq.util.LongSequenceGenerator;
import org.apache.activemq.util.ServiceStopper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


/**
 * Routes Broker operations to the correct messaging regions for processing.
 * 
 * @version $Revision$
 */
public class RegionBroker implements Broker {
    private static final Log log = LogFactory.getLog(RegionBroker.class);
    private static final IdGenerator brokerIdGenerator = new IdGenerator();

    private final Region queueRegion;
    private final Region topicRegion;
    private final Region tempQueueRegion;
    private final Region tempTopicRegion;
    private BrokerService brokerService;
    private boolean started = false;
    private boolean keepDurableSubsActive=false;
    
    protected final DestinationStatistics destinationStatistics = new DestinationStatistics();
    
    private final CopyOnWriteArrayList connections = new CopyOnWriteArrayList();
    private final HashMap destinations = new HashMap();
    private final CopyOnWriteArrayList brokerInfos = new CopyOnWriteArrayList();

    private final LongSequenceGenerator sequenceGenerator = new LongSequenceGenerator();    
    private BrokerId brokerId;
    private String brokerName;
    private Map<String, ConnectionContext> clientIdSet = new HashMap<String, ConnectionContext>(); // we will synchronize access
    private final DestinationInterceptor destinationInterceptor;
    private ConnectionContext adminConnectionContext;
    protected DestinationFactory destinationFactory;
    protected final ConcurrentHashMap connectionStates = new ConcurrentHashMap();
   
    
        
    public RegionBroker(BrokerService brokerService,TaskRunnerFactory taskRunnerFactory, UsageManager memoryManager, DestinationFactory destinationFactory, DestinationInterceptor destinationInterceptor) throws IOException {
        this.brokerService = brokerService;
        if (destinationFactory == null) {
            throw new IllegalArgumentException("null destinationFactory");
        }
        this.sequenceGenerator.setLastSequenceId( destinationFactory.getLastMessageBrokerSequenceId() );
        this.destinationFactory = destinationFactory;
        queueRegion = createQueueRegion(memoryManager, taskRunnerFactory, destinationFactory);
        topicRegion = createTopicRegion(memoryManager, taskRunnerFactory, destinationFactory);
        this.destinationInterceptor = destinationInterceptor;
        tempQueueRegion = createTempQueueRegion(memoryManager, taskRunnerFactory, destinationFactory);
        tempTopicRegion = createTempTopicRegion(memoryManager, taskRunnerFactory, destinationFactory);        
    }
    
    public Map getDestinationMap() {
        Map answer = getQueueRegion().getDestinationMap();
        answer.putAll(getTopicRegion().getDestinationMap());
        return answer;
    }

    public Set getDestinations(ActiveMQDestination destination) {
        switch(destination.getDestinationType()) {
        case ActiveMQDestination.QUEUE_TYPE:
            return queueRegion.getDestinations(destination);
        case ActiveMQDestination.TOPIC_TYPE:
            return topicRegion.getDestinations(destination);
        case ActiveMQDestination.TEMP_QUEUE_TYPE:
            return tempQueueRegion.getDestinations(destination);
        case ActiveMQDestination.TEMP_TOPIC_TYPE:
            return tempTopicRegion.getDestinations(destination);
        default:
            return Collections.EMPTY_SET;
        }
    }

    public Broker getAdaptor(Class type){
        if (type.isInstance(this)){
            return this;
        }
        return null;
    }

    public Region getQueueRegion() {
        return queueRegion;
    }

    public Region getTempQueueRegion() {
        return tempQueueRegion;
    }

    public Region getTempTopicRegion() {
        return tempTopicRegion;
    }

    public Region getTopicRegion() {
        return topicRegion;
    }

    protected Region createTempTopicRegion(UsageManager memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
        return new TempTopicRegion(this,destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
    }

    protected Region createTempQueueRegion(UsageManager memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
        return new TempQueueRegion(this,destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
    }

    protected Region createTopicRegion(UsageManager memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
        return new TopicRegion(this,destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
    }

    protected Region createQueueRegion(UsageManager memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
        return new QueueRegion(this,destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
    }
    
    private static PersistenceAdapter createDefaultPersistenceAdapter(UsageManager memoryManager) throws IOException {
        return new MemoryPersistenceAdapter();
    }
    
    
    public void start() throws Exception {
        ((TopicRegion)topicRegion).setKeepDurableSubsActive(keepDurableSubsActive);
        started = true;
        queueRegion.start();
        topicRegion.start();
        tempQueueRegion.start();
        tempTopicRegion.start();
    }

    public void stop() throws Exception {
        started = false;
        ServiceStopper ss = new ServiceStopper();
        doStop(ss);
        ss.throwFirstException();
    }
    
    public PolicyMap getDestinationPolicy(){
        return brokerService != null ? brokerService.getDestinationPolicy() : null;
    }

    public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
        String clientId = info.getClientId();
        if (clientId == null) {
            throw new InvalidClientIDException("No clientID specified for connection request");
        }
        synchronized (clientIdSet ) {
        	ConnectionContext oldContext = clientIdSet.get(clientId);
            if (oldContext!=null) {
            	throw new InvalidClientIDException("Broker: " + getBrokerName() + " - Client: " + clientId + " already connected from "+oldContext.getConnection().getRemoteAddress());
            }
            else {
                clientIdSet.put(clientId, context);
            }
        }

        connections.add(context.getConnection());
    }

    public void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error) throws Exception {
        String clientId = info.getClientId();
        if (clientId == null) {
            throw new InvalidClientIDException("No clientID specified for connection disconnect request");
        }
        synchronized (clientIdSet) {
            ConnectionContext oldValue = clientIdSet.get(clientId);
            // we may be removing the duplicate connection, not the first connection to be created
            // so lets check that their connection IDs are the same
            if (oldValue == context ) {
                if (isEqual(oldValue.getConnectionId(), info.getConnectionId())) {
                    clientIdSet.remove(clientId);
                }
            }
        }
        connections.remove(context.getConnection());
    }

    protected boolean isEqual(ConnectionId connectionId, ConnectionId connectionId2) {
        return connectionId == connectionId2 || (connectionId != null && connectionId.equals(connectionId2));
    }

    public Connection[] getClients() throws Exception {
        ArrayList l = new ArrayList(connections);
        Connection rc[] = new Connection[l.size()];
        l.toArray(rc);
        return rc;
    }

    public Destination addDestination(ConnectionContext context, ActiveMQDestination destination) throws Exception {
  
    	Destination answer;
    	synchronized(destinations) {
    		answer = (Destination) destinations.get(destination);
            if( answer!=null )
            	return answer;
    	}
    	
        switch(destination.getDestinationType()) {
        case ActiveMQDestination.QUEUE_TYPE:
            answer  = queueRegion.addDestination(context, destination);
            break;
        case ActiveMQDestination.TOPIC_TYPE:
            answer = topicRegion.addDestination(context, destination);
            break;
        case ActiveMQDestination.TEMP_QUEUE_TYPE:
            answer = tempQueueRegion.addDestination(context, destination);
            break;
        case ActiveMQDestination.TEMP_TOPIC_TYPE:
            answer = tempTopicRegion.addDestination(context, destination);
            break;
        default:
            throw createUnknownDestinationTypeException(destination);
        }

    	synchronized(destinations) {
            destinations.put(destination, answer);
            return answer;
    	}
    }

    public void removeDestination(ConnectionContext context,ActiveMQDestination destination,long timeout) throws Exception{
    	synchronized(destinations) {
	        if( destinations.remove(destination)!=null ){
	            switch(destination.getDestinationType()){
	            case ActiveMQDestination.QUEUE_TYPE:
	                queueRegion.removeDestination(context,destination,timeout);
	                break;
	            case ActiveMQDestination.TOPIC_TYPE:
	                topicRegion.removeDestination(context,destination,timeout);
	                break;
	            case ActiveMQDestination.TEMP_QUEUE_TYPE:
	                tempQueueRegion.removeDestination(context,destination,timeout);
	                break;
	            case ActiveMQDestination.TEMP_TOPIC_TYPE:
	                tempTopicRegion.removeDestination(context,destination,timeout);
	                break;
	            default:
	                throw createUnknownDestinationTypeException(destination);
	            }
	        }
    	}
    }
    
    public void addDestinationInfo(ConnectionContext context,DestinationInfo info) throws Exception{
        addDestination(context,info.getDestination());
        
    }

    public void removeDestinationInfo(ConnectionContext context,DestinationInfo info) throws Exception{
        removeDestination(context,info.getDestination(), info.getTimeout());
        
    }

    public ActiveMQDestination[] getDestinations() throws Exception {
    	ArrayList l;
    	synchronized(destinations) {
	        l = new ArrayList(destinations.values());
    	}
        ActiveMQDestination rc[] = new ActiveMQDestination[l.size()];
        l.toArray(rc);
        return rc;
    }


    public void addSession(ConnectionContext context, SessionInfo info) throws Exception {
    }

    public void removeSession(ConnectionContext context, SessionInfo info) throws Exception {<commit_msg>Fix for AMQ-1344 - Deadlock can occur when cerating and deleting many destinations concurrently<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.region;

import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

import javax.jms.InvalidClientIDException;
import javax.jms.JMSException;

import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.Connection;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.ConsumerBrokerExchange;
import org.apache.activemq.broker.ProducerBrokerExchange;
import org.apache.activemq.broker.region.policy.DeadLetterStrategy;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.BrokerId;
import org.apache.activemq.command.BrokerInfo;
import org.apache.activemq.command.ConnectionId;
import org.apache.activemq.command.ConnectionInfo;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.DestinationInfo;
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.MessagePull;
import org.apache.activemq.command.ProducerInfo;
import org.apache.activemq.command.RemoveSubscriptionInfo;
import org.apache.activemq.command.Response;
import org.apache.activemq.command.SessionInfo;
import org.apache.activemq.command.TransactionId;
import org.apache.activemq.kaha.Store;
import org.apache.activemq.memory.UsageManager;
import org.apache.activemq.store.PersistenceAdapter;
import org.apache.activemq.store.memory.MemoryPersistenceAdapter;
import org.apache.activemq.thread.TaskRunnerFactory;
import org.apache.activemq.util.BrokerSupport;
import org.apache.activemq.util.IdGenerator;
import org.apache.activemq.util.LongSequenceGenerator;
import org.apache.activemq.util.ServiceStopper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


/**
 * Routes Broker operations to the correct messaging regions for processing.
 * 
 * @version $Revision$
 */
public class RegionBroker implements Broker {
    private static final Log log = LogFactory.getLog(RegionBroker.class);
    private static final IdGenerator brokerIdGenerator = new IdGenerator();

    private final Region queueRegion;
    private final Region topicRegion;
    private final Region tempQueueRegion;
    private final Region tempTopicRegion;
    private BrokerService brokerService;
    private boolean started = false;
    private boolean keepDurableSubsActive=false;
    
    protected final DestinationStatistics destinationStatistics = new DestinationStatistics();
    
    private final CopyOnWriteArrayList connections = new CopyOnWriteArrayList();
    private final Map destinations = new ConcurrentHashMap();
    private final CopyOnWriteArrayList brokerInfos = new CopyOnWriteArrayList();

    private final LongSequenceGenerator sequenceGenerator = new LongSequenceGenerator();    
    private BrokerId brokerId;
    private String brokerName;
    private Map<String, ConnectionContext> clientIdSet = new HashMap<String, ConnectionContext>(); // we will synchronize access
    private final DestinationInterceptor destinationInterceptor;
    private ConnectionContext adminConnectionContext;
    protected DestinationFactory destinationFactory;
    protected final ConcurrentHashMap connectionStates = new ConcurrentHashMap();
   
    
        
    public RegionBroker(BrokerService brokerService,TaskRunnerFactory taskRunnerFactory, UsageManager memoryManager, DestinationFactory destinationFactory, DestinationInterceptor destinationInterceptor) throws IOException {
        this.brokerService = brokerService;
        if (destinationFactory == null) {
            throw new IllegalArgumentException("null destinationFactory");
        }
        this.sequenceGenerator.setLastSequenceId( destinationFactory.getLastMessageBrokerSequenceId() );
        this.destinationFactory = destinationFactory;
        queueRegion = createQueueRegion(memoryManager, taskRunnerFactory, destinationFactory);
        topicRegion = createTopicRegion(memoryManager, taskRunnerFactory, destinationFactory);
        this.destinationInterceptor = destinationInterceptor;
        tempQueueRegion = createTempQueueRegion(memoryManager, taskRunnerFactory, destinationFactory);
        tempTopicRegion = createTempTopicRegion(memoryManager, taskRunnerFactory, destinationFactory);        
    }
    
    public Map getDestinationMap() {
        Map answer = getQueueRegion().getDestinationMap();
        answer.putAll(getTopicRegion().getDestinationMap());
        return answer;
    }

    public Set getDestinations(ActiveMQDestination destination) {
        switch(destination.getDestinationType()) {
        case ActiveMQDestination.QUEUE_TYPE:
            return queueRegion.getDestinations(destination);
        case ActiveMQDestination.TOPIC_TYPE:
            return topicRegion.getDestinations(destination);
        case ActiveMQDestination.TEMP_QUEUE_TYPE:
            return tempQueueRegion.getDestinations(destination);
        case ActiveMQDestination.TEMP_TOPIC_TYPE:
            return tempTopicRegion.getDestinations(destination);
        default:
            return Collections.EMPTY_SET;
        }
    }

    public Broker getAdaptor(Class type){
        if (type.isInstance(this)){
            return this;
        }
        return null;
    }

    public Region getQueueRegion() {
        return queueRegion;
    }

    public Region getTempQueueRegion() {
        return tempQueueRegion;
    }

    public Region getTempTopicRegion() {
        return tempTopicRegion;
    }

    public Region getTopicRegion() {
        return topicRegion;
    }

    protected Region createTempTopicRegion(UsageManager memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
        return new TempTopicRegion(this,destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
    }

    protected Region createTempQueueRegion(UsageManager memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
        return new TempQueueRegion(this,destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
    }

    protected Region createTopicRegion(UsageManager memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
        return new TopicRegion(this,destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
    }

    protected Region createQueueRegion(UsageManager memoryManager, TaskRunnerFactory taskRunnerFactory, DestinationFactory destinationFactory) {
        return new QueueRegion(this,destinationStatistics, memoryManager, taskRunnerFactory, destinationFactory);
    }
    
    private static PersistenceAdapter createDefaultPersistenceAdapter(UsageManager memoryManager) throws IOException {
        return new MemoryPersistenceAdapter();
    }
    
    
    public void start() throws Exception {
        ((TopicRegion)topicRegion).setKeepDurableSubsActive(keepDurableSubsActive);
        started = true;
        queueRegion.start();
        topicRegion.start();
        tempQueueRegion.start();
        tempTopicRegion.start();
    }

    public void stop() throws Exception {
        started = false;
        ServiceStopper ss = new ServiceStopper();
        doStop(ss);
        ss.throwFirstException();
    }
    
    public PolicyMap getDestinationPolicy(){
        return brokerService != null ? brokerService.getDestinationPolicy() : null;
    }

    public void addConnection(ConnectionContext context, ConnectionInfo info) throws Exception {
        String clientId = info.getClientId();
        if (clientId == null) {
            throw new InvalidClientIDException("No clientID specified for connection request");
        }
        synchronized (clientIdSet ) {
        	ConnectionContext oldContext = clientIdSet.get(clientId);
            if (oldContext!=null) {
            	throw new InvalidClientIDException("Broker: " + getBrokerName() + " - Client: " + clientId + " already connected from "+oldContext.getConnection().getRemoteAddress());
            }
            else {
                clientIdSet.put(clientId, context);
            }
        }

        connections.add(context.getConnection());
    }

    public void removeConnection(ConnectionContext context, ConnectionInfo info, Throwable error) throws Exception {
        String clientId = info.getClientId();
        if (clientId == null) {
            throw new InvalidClientIDException("No clientID specified for connection disconnect request");
        }
        synchronized (clientIdSet) {
            ConnectionContext oldValue = clientIdSet.get(clientId);
            // we may be removing the duplicate connection, not the first connection to be created
            // so lets check that their connection IDs are the same
            if (oldValue == context ) {
                if (isEqual(oldValue.getConnectionId(), info.getConnectionId())) {
                    clientIdSet.remove(clientId);
                }
            }
        }
        connections.remove(context.getConnection());
    }

    protected boolean isEqual(ConnectionId connectionId, ConnectionId connectionId2) {
        return connectionId == connectionId2 || (connectionId != null && connectionId.equals(connectionId2));
    }

    public Connection[] getClients() throws Exception {
        ArrayList l = new ArrayList(connections);
        Connection rc[] = new Connection[l.size()];
        l.toArray(rc);
        return rc;
    }

    public Destination addDestination(ConnectionContext context, ActiveMQDestination destination) throws Exception {

		Destination answer;

		answer = (Destination) destinations.get(destination);
		if (answer != null)
			return answer;

		switch (destination.getDestinationType()) {
		case ActiveMQDestination.QUEUE_TYPE:
			answer = queueRegion.addDestination(context, destination);
			break;
		case ActiveMQDestination.TOPIC_TYPE:
			answer = topicRegion.addDestination(context, destination);
			break;
		case ActiveMQDestination.TEMP_QUEUE_TYPE:
			answer = tempQueueRegion.addDestination(context, destination);
			break;
		case ActiveMQDestination.TEMP_TOPIC_TYPE:
			answer = tempTopicRegion.addDestination(context, destination);
			break;
		default:
			throw createUnknownDestinationTypeException(destination);
		}

		destinations.put(destination, answer);
		return answer;

	}

	public void removeDestination(ConnectionContext context,
			ActiveMQDestination destination, long timeout) throws Exception {

		if (destinations.remove(destination) != null) {
			switch (destination.getDestinationType()) {
			case ActiveMQDestination.QUEUE_TYPE:
				queueRegion.removeDestination(context, destination, timeout);
				break;
			case ActiveMQDestination.TOPIC_TYPE:
				topicRegion.removeDestination(context, destination, timeout);
				break;
			case ActiveMQDestination.TEMP_QUEUE_TYPE:
				tempQueueRegion
						.removeDestination(context, destination, timeout);
				break;
			case ActiveMQDestination.TEMP_TOPIC_TYPE:
				tempTopicRegion
						.removeDestination(context, destination, timeout);
				break;
			default:
				throw createUnknownDestinationTypeException(destination);
			}
		}

	}
    
    public void addDestinationInfo(ConnectionContext context,DestinationInfo info) throws Exception{
        addDestination(context,info.getDestination());
        
    }

    public void removeDestinationInfo(ConnectionContext context,DestinationInfo info) throws Exception{
        removeDestination(context,info.getDestination(), info.getTimeout());
        
    }

    public ActiveMQDestination[] getDestinations() throws Exception {
		ArrayList l;

		l = new ArrayList(destinations.values());

		ActiveMQDestination rc[] = new ActiveMQDestination[l.size()];
		l.toArray(rc);
		return rc;
	}


    public void addSession(ConnectionContext context, SessionInfo info) throws Exception {
    }

    public void removeSession(ConnectionContext context, SessionInfo info) throws Exception {
{'fraction_non_alphanumeric': 0.07459104168974558, 'fraction_numerical': 0.0006646726487205051, 'mean_word_length': 3.758741873133017, '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': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1568680', 'n_tokens_mistral': 6662, 'n_tokens_neox': 6531, 'n_words': 1622}