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

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.activemq.command.DiscoveryEvent;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFactory;
import org.apache.activemq.transport.discovery.DiscoveryAgent;
import org.apache.activemq.transport.discovery.DiscoveryAgentFactory;
import org.apache.activemq.transport.discovery.DiscoveryListener;
import org.apache.activemq.util.ServiceStopper;
import org.apache.activemq.util.ServiceSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * A network connector which uses a discovery agent to detect the remote brokers
 * available and setup a connection to each available remote broker
 * 
 * @org.apache.xbean.XBean element="networkConnector"
 * @version $Revision$
 */
public class DiscoveryNetworkConnector extends NetworkConnector implements DiscoveryListener {
    private static final Log LOG = LogFactory.getLog(DiscoveryNetworkConnector.class);

    private DiscoveryAgent discoveryAgent;
    private ConcurrentHashMap<URI, NetworkBridge> bridges = new ConcurrentHashMap<URI, NetworkBridge>();

    public DiscoveryNetworkConnector() {
    }

    public DiscoveryNetworkConnector(URI discoveryURI) throws IOException {
        setUri(discoveryURI);
    }

    public void setUri(URI discoveryURI) throws IOException {
        setDiscoveryAgent(DiscoveryAgentFactory.createDiscoveryAgent(discoveryURI));
    }

    public void onServiceAdd(DiscoveryEvent event) {
        String localURIName = localURI.getScheme() + "://" + localURI.getHost();
        // Ignore events once we start stopping.
        if (serviceSupport.isStopped() || serviceSupport.isStopping()) {
            return;
        }
        String url = event.getServiceName();
        if (url != null) {
            URI uri;
            try {
                uri = new URI(url);
            } catch (URISyntaxException e) {
                LOG.warn("Could not connect to remote URI: " + url + " due to bad URI syntax: " + e, e);
                return;
            }
            // Should we try to connect to that URI?
            if (bridges.containsKey(uri) || localURI.equals(uri) || (connectionFilter != null && !connectionFilter.connectTo(uri))) {
                return;
            }
            URI connectUri = uri;
            LOG.info("Establishing network connection between from " + localURIName + " to " + connectUri);
            Transport remoteTransport;
            try {
                remoteTransport = TransportFactory.connect(connectUri);
            } catch (Exception e) {
                LOG.warn("Could not connect to remote URI: " + localURIName + ": " + e.getMessage());
                LOG.debug("Connection failure exception: " + e, e);
                return;
            }
            Transport localTransport;
            try {
                localTransport = createLocalTransport();
            } catch (Exception e) {
                ServiceSupport.dispose(remoteTransport);
                LOG.warn("Could not connect to local URI: " + localURIName + ": " + e.getMessage());
                LOG.debug("Connection failure exception: " + e, e);
                return;
            }
            NetworkBridge bridge = createBridge(localTransport, remoteTransport, event);
            bridges.put(uri, bridge);
            try {
                bridge.start();
            } catch (Exception e) {<commit_msg>Add some info logging so you can check to see if the discovery agent is generating duplicate onServiceAdd events<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.network;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.activemq.command.DiscoveryEvent;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFactory;
import org.apache.activemq.transport.discovery.DiscoveryAgent;
import org.apache.activemq.transport.discovery.DiscoveryAgentFactory;
import org.apache.activemq.transport.discovery.DiscoveryListener;
import org.apache.activemq.util.ServiceStopper;
import org.apache.activemq.util.ServiceSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * A network connector which uses a discovery agent to detect the remote brokers
 * available and setup a connection to each available remote broker
 * 
 * @org.apache.xbean.XBean element="networkConnector"
 * @version $Revision$
 */
public class DiscoveryNetworkConnector extends NetworkConnector implements DiscoveryListener {
    private static final Log LOG = LogFactory.getLog(DiscoveryNetworkConnector.class);

    private DiscoveryAgent discoveryAgent;
    private ConcurrentHashMap<URI, NetworkBridge> bridges = new ConcurrentHashMap<URI, NetworkBridge>();

    public DiscoveryNetworkConnector() {
    }

    public DiscoveryNetworkConnector(URI discoveryURI) throws IOException {
        setUri(discoveryURI);
    }

    public void setUri(URI discoveryURI) throws IOException {
        setDiscoveryAgent(DiscoveryAgentFactory.createDiscoveryAgent(discoveryURI));
    }

    public void onServiceAdd(DiscoveryEvent event) {
        String localURIName = localURI.getScheme() + "://" + localURI.getHost();
        // Ignore events once we start stopping.
        if (serviceSupport.isStopped() || serviceSupport.isStopping()) {
            return;
        }
        String url = event.getServiceName();
        if (url != null) {
            URI uri;
            try {
                uri = new URI(url);
            } catch (URISyntaxException e) {
                LOG.warn("Could not connect to remote URI: " + url + " due to bad URI syntax: " + e, e);
                return;
            }
            // Should we try to connect to that URI?
            if( bridges.containsKey(uri) ) {
                LOG.debug("Discovery agent generated a duplicate onServiceAdd event for: "+uri );
                return;
            }
            if ( localURI.equals(uri) || (connectionFilter != null && !connectionFilter.connectTo(uri))) {
                return;
            }
            URI connectUri = uri;
            LOG.info("Establishing network connection between from " + localURIName + " to " + connectUri);
            Transport remoteTransport;
            try {
                remoteTransport = TransportFactory.connect(connectUri);
            } catch (Exception e) {
                LOG.warn("Could not connect to remote URI: " + localURIName + ": " + e.getMessage());
                LOG.debug("Connection failure exception: " + e, e);
                return;
            }
            Transport localTransport;
            try {
                localTransport = createLocalTransport();
            } catch (Exception e) {
                ServiceSupport.dispose(remoteTransport);
                LOG.warn("Could not connect to local URI: " + localURIName + ": " + e.getMessage());
                LOG.debug("Connection failure exception: " + e, e);
                return;
            }
            NetworkBridge bridge = createBridge(localTransport, remoteTransport, event);
            bridges.put(uri, bridge);
            try {
                bridge.start();
            } catch (Exception e) {
{'fraction_non_alphanumeric': 0.0871423873725748, 'fraction_numerical': 0.0008769045270196207, 'mean_word_length': 2.737812371978697, 'pattern_counts': {'":': 6, '<': 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': '2019888', 'n_tokens_mistral': 2270, 'n_tokens_neox': 2210, 'n_words': 711}