<commit_before>
package org.jgroups.protocols;

import java.io.DataInputStream;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

import org.jgroups.Address;
import org.jgroups.Event;
import org.jgroups.Global;
import org.jgroups.Message;
import org.jgroups.PhysicalAddress;
import org.jgroups.annotations.Experimental;
import org.jgroups.annotations.Property;
import org.jgroups.stack.GossipData;
import org.jgroups.stack.GossipRouter;
import org.jgroups.stack.RouterStubManager;
import org.jgroups.stack.IpAddress;
import org.jgroups.stack.Protocol;
import org.jgroups.stack.RouterStub;
import org.jgroups.util.Buffer;
import org.jgroups.util.ExposedByteArrayOutputStream;
import org.jgroups.util.ExposedDataOutputStream;
import org.jgroups.util.Util;

/**
 * Replacement for UDP. Instead of sending packets via UDP, a TCP connection is opened to a Router
 * (using the RouterStub client-side stub), the IP address/port of which was given using channel
 * properties <code>router_host</code> and <code>router_port</code>. All outgoing traffic is sent
 * via this TCP socket to the Router which distributes it to all connected TUNNELs in this group.
 * Incoming traffic received from Router will simply be passed up the stack.
 * 
 * <p>
 * A TUNNEL layer can be used to penetrate a firewall, most firewalls allow creating TCP connections
 * to the outside world, however, they do not permit outside hosts to initiate a TCP connection to a
 * host inside the firewall. Therefore, the connection created by the inside host is reused by
 * Router to send traffic from an outside host to a host inside the firewall.
 * 
 * @author <NAME>
 * @author <NAME>
 */
@Experimental
public class TUNNEL extends TP {

    /*
    * ----------------------------------------- Properties
    * --------------------------------------------------
    */

    @Property(description = "Interval in msec to attempt connecting back to router in case of torn connection. Default is 5000 msec")
    private long reconnect_interval = 5000;

    @Property(description="Should TCP no delay flag be turned on")
    boolean tcp_nodelay=false;

    /*
    * --------------------------------------------- Fields
    * ------------------------------------------------------
    */

   private final List<InetSocketAddress> gossip_router_hosts = new ArrayList<InetSocketAddress>();

   private TUNNELPolicy tunnel_policy = new DefaultTUNNELPolicy();

   private DatagramSocket sock;
   
   private volatile RouterStubManager stubManager;

   public TUNNEL() {
   }

    public boolean supportsMulticasting() {
        return false;
    }

    @Property(description="A comma-separated list of GossipRouter hosts, e.g. HostA[12001],HostB[12001]")
   public void setGossipRouterHosts(String hosts) throws UnknownHostException {
      gossip_router_hosts.clear();
      // if we get passed value of List<SocketAddress>#toString() we have to strip []
      if (hosts.startsWith("[") && hosts.endsWith("]")) {
         hosts = hosts.substring(1, hosts.length() - 1);
      }
      gossip_router_hosts.addAll(Util.parseCommaDelimitedHosts2(hosts, 1));
   }

   public String toString() {
      return "TUNNEL";
   }

   public long getReconnectInterval() {
      return reconnect_interval;
   }

   public void setReconnectInterval(long reconnect_interval) {
      this.reconnect_interval = reconnect_interval;
   }

   /*------------------------------ Protocol interface ------------------------------ */

   public synchronized void setTUNNELPolicy(TUNNELPolicy policy) {
      if (policy == null)
         throw new IllegalArgumentException("Tunnel policy has to be non null");
      tunnel_policy = policy;
   }

    public void init() throws Exception {
        super.init();

        if(enable_bundling) {
            log.warn("bundling is currently not supported by TUNNEL; bundling is disabled");
            enable_bundling=false;
        }

        if (timer == null)
            throw new Exception("timer cannot be retrieved from protocol stack");
      
        // Postpone TUNNEL and shared transport until 3.0 timeframe
        // TODO [JGRP-1194] - Revisit implementation of TUNNEL and shared transport
        if (isSingleton()) {
            throw new Exception("TUNNEL and shared transport mode are not supported!");
        }

        if (log.isDebugEnabled())
            log.debug("GossipRouters are:" + gossip_router_hosts.toString());
        
        stubManager = RouterStubManager.emptyGossipClientStubManager(this);
        sock = getSocketFactory().createDatagramSocket(Global.TUNNEL_UCAST_SOCK, bind_port, bind_addr);
        
        // loopback turned on is mandatory
        loopback = true;
    }
    
    public void destroy() {        
        stubManager.destroyStubs();
        super.destroy();
    }

   private void disconnectStub(String group, Address addr) {
      stubManager.disconnectStubs();
   }

   public Object handleDownEvent(Event evt) {
      Object retEvent = super.handleDownEvent(evt);
      switch (evt.getType()) {
         case Event.CONNECT:
         case Event.CONNECT_WITH_STATE_TRANSFER:
         case Event.CONNECT_USE_FLUSH:
         case Event.CONNECT_WITH_STATE_TRANSFER_USE_FLUSH:
             String group=(String)evt.getArg();
             Address local= null;
             if(!isSingleton()) {
                 local = local_addr;                 
             } else {
                 // TODO [JGRP-1194] - Revisit implementation of TUNNEL and shared transport   
                 ProtocolAdapter adapter = ProtocolAdapter.thread_local.get();
                 local = adapter.local_addr;
             }
             
             stubManager = new TUNNELStubManager(this,group,local,getReconnectInterval());
             for (InetSocketAddress gr : gossip_router_hosts) {
                 RouterStub stub = stubManager.createAndRegisterStub(gr.getHostName(), gr.getPort(), bind_addr);
                 stub.setTcpNoDelay(tcp_nodelay);           
              }  
             PhysicalAddress physical_addr=(PhysicalAddress)down(new Event(Event.GET_PHYSICAL_ADDRESS, local));
             List<PhysicalAddress> physical_addrs=Arrays.asList(physical_addr);
             String logical_name=org.jgroups.util.UUID.get(local);
             List<RouterStub> stubs = stubManager.getStubs();
             tunnel_policy.connect(stubs, group, local, logical_name, physical_addrs);
            break;

         case Event.DISCONNECT:
             if(!isSingleton()) {
                 local = local_addr;        
                 group = channel_name;
             } else {
                 // TODO [JGRP-1194] - Revisit implementation of TUNNEL and shared transport
                 ProtocolAdapter adapter = ProtocolAdapter.thread_local.get();
                 local = adapter.local_addr;
                 group = adapter.cluster_name;
             }
             disconnectStub(group,local);
            break;
      }
      return retEvent;
   }
    private class TUNNELStubManager extends RouterStubManager {
<commit_msg>JGRP-1359 - TUNNEL:Destroy StubManager before creating a new one <commit_after>
package org.jgroups.protocols;

import java.io.DataInputStream;
import java.net.DatagramSocket;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;

import org.jgroups.Address;
import org.jgroups.Event;
import org.jgroups.Global;
import org.jgroups.Message;
import org.jgroups.PhysicalAddress;
import org.jgroups.annotations.Experimental;
import org.jgroups.annotations.Property;
import org.jgroups.stack.GossipData;
import org.jgroups.stack.GossipRouter;
import org.jgroups.stack.RouterStubManager;
import org.jgroups.stack.IpAddress;
import org.jgroups.stack.Protocol;
import org.jgroups.stack.RouterStub;
import org.jgroups.util.Buffer;
import org.jgroups.util.ExposedByteArrayOutputStream;
import org.jgroups.util.ExposedDataOutputStream;
import org.jgroups.util.Util;

/**
 * Replacement for UDP. Instead of sending packets via UDP, a TCP connection is opened to a Router
 * (using the RouterStub client-side stub), the IP address/port of which was given using channel
 * properties <code>router_host</code> and <code>router_port</code>. All outgoing traffic is sent
 * via this TCP socket to the Router which distributes it to all connected TUNNELs in this group.
 * Incoming traffic received from Router will simply be passed up the stack.
 * 
 * <p>
 * A TUNNEL layer can be used to penetrate a firewall, most firewalls allow creating TCP connections
 * to the outside world, however, they do not permit outside hosts to initiate a TCP connection to a
 * host inside the firewall. Therefore, the connection created by the inside host is reused by
 * Router to send traffic from an outside host to a host inside the firewall.
 * 
 * @author <NAME>
 * @author <NAME>
 */
@Experimental
public class TUNNEL extends TP {

    /*
    * ----------------------------------------- Properties
    * --------------------------------------------------
    */

    @Property(description = "Interval in msec to attempt connecting back to router in case of torn connection. Default is 5000 msec")
    private long reconnect_interval = 5000;

    @Property(description="Should TCP no delay flag be turned on")
    boolean tcp_nodelay=false;

    /*
    * --------------------------------------------- Fields
    * ------------------------------------------------------
    */

   private final List<InetSocketAddress> gossip_router_hosts = new ArrayList<InetSocketAddress>();

   private TUNNELPolicy tunnel_policy = new DefaultTUNNELPolicy();

   private DatagramSocket sock;
   
   private volatile RouterStubManager stubManager;

   public TUNNEL() {
   }

    public boolean supportsMulticasting() {
        return false;
    }

    @Property(description="A comma-separated list of GossipRouter hosts, e.g. HostA[12001],HostB[12001]")
   public void setGossipRouterHosts(String hosts) throws UnknownHostException {
      gossip_router_hosts.clear();
      // if we get passed value of List<SocketAddress>#toString() we have to strip []
      if (hosts.startsWith("[") && hosts.endsWith("]")) {
         hosts = hosts.substring(1, hosts.length() - 1);
      }
      gossip_router_hosts.addAll(Util.parseCommaDelimitedHosts2(hosts, 1));
   }

   public String toString() {
      return "TUNNEL";
   }

   public long getReconnectInterval() {
      return reconnect_interval;
   }

   public void setReconnectInterval(long reconnect_interval) {
      this.reconnect_interval = reconnect_interval;
   }

   /*------------------------------ Protocol interface ------------------------------ */

   public synchronized void setTUNNELPolicy(TUNNELPolicy policy) {
      if (policy == null)
         throw new IllegalArgumentException("Tunnel policy has to be non null");
      tunnel_policy = policy;
   }

    public void init() throws Exception {
        super.init();

        if(enable_bundling) {
            log.warn("bundling is currently not supported by TUNNEL; bundling is disabled");
            enable_bundling=false;
        }

        if (timer == null)
            throw new Exception("timer cannot be retrieved from protocol stack");
      
        // Postpone TUNNEL and shared transport until 3.0 timeframe
        // TODO [JGRP-1194] - Revisit implementation of TUNNEL and shared transport
        if (isSingleton()) {
            throw new Exception("TUNNEL and shared transport mode are not supported!");
        }

        if (log.isDebugEnabled())
            log.debug("GossipRouters are:" + gossip_router_hosts.toString());
        
        stubManager = RouterStubManager.emptyGossipClientStubManager(this);
        sock = getSocketFactory().createDatagramSocket(Global.TUNNEL_UCAST_SOCK, bind_port, bind_addr);
        
        // loopback turned on is mandatory
        loopback = true;
    }
    
    public void destroy() {        
        stubManager.destroyStubs();
        super.destroy();
    }

   private void disconnectStub(String group, Address addr) {
      stubManager.disconnectStubs();
   }

   public Object handleDownEvent(Event evt) {
      Object retEvent = super.handleDownEvent(evt);
      switch (evt.getType()) {
         case Event.CONNECT:
         case Event.CONNECT_WITH_STATE_TRANSFER:
         case Event.CONNECT_USE_FLUSH:
         case Event.CONNECT_WITH_STATE_TRANSFER_USE_FLUSH:
             String group=(String)evt.getArg();
             Address local= null;
             if(!isSingleton()) {
                 local = local_addr;                 
             } else {
                 // TODO [JGRP-1194] - Revisit implementation of TUNNEL and shared transport   
                 ProtocolAdapter adapter = ProtocolAdapter.thread_local.get();
                 local = adapter.local_addr;
             }
             
             if(stubManager != null) {
                stubManager.destroyStubs();
             }
             stubManager = new TUNNELStubManager(this,group,local,getReconnectInterval());
             for (InetSocketAddress gr : gossip_router_hosts) {
                 RouterStub stub = stubManager.createAndRegisterStub(gr.getHostName(), gr.getPort(), bind_addr);
                 stub.setTcpNoDelay(tcp_nodelay);           
              }  
             PhysicalAddress physical_addr=(PhysicalAddress)down(new Event(Event.GET_PHYSICAL_ADDRESS, local));
             List<PhysicalAddress> physical_addrs=Arrays.asList(physical_addr);
             String logical_name=org.jgroups.util.UUID.get(local);
             List<RouterStub> stubs = stubManager.getStubs();
             tunnel_policy.connect(stubs, group, local, logical_name, physical_addrs);
            break;

         case Event.DISCONNECT:
             if(!isSingleton()) {
                 local = local_addr;        
                 group = channel_name;
             } else {
                 // TODO [JGRP-1194] - Revisit implementation of TUNNEL and shared transport
                 ProtocolAdapter adapter = ProtocolAdapter.thread_local.get();
                 local = adapter.local_addr;
                 group = adapter.cluster_name;
             }
             disconnectStub(group,local);
            break;
      }
      return retEvent;
   }
    private class TUNNELStubManager extends RouterStubManager {

{'fraction_non_alphanumeric': 0.11535822124622563, 'fraction_numerical': 0.00521548174581389, 'mean_word_length': 2.410484437163585, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '5699291', 'n_tokens_mistral': 3911, 'n_tokens_neox': 3849, 'n_words': 1053}