content: /*
   * 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.transport.udp;
  
  import java.io.EOFException;
  import java.io.IOException;
  import java.net.BindException;
  import java.net.DatagramSocket;
  import java.net.InetSocketAddress;
  import java.net.SocketAddress;
  import java.net.SocketException;
  import java.net.URI;
  import java.net.UnknownHostException;
  import java.nio.channels.AsynchronousCloseException;
  import java.nio.channels.DatagramChannel;
  import java.security.cert.X509Certificate;
  
  import org.apache.activemq.Service;
  import org.apache.activemq.command.Command;
  import org.apache.activemq.command.Endpoint;
  import org.apache.activemq.openwire.OpenWireFormat;
  import org.apache.activemq.transport.Transport;
  import org.apache.activemq.transport.TransportThreadSupport;
  import org.apache.activemq.transport.reliable.ExceptionIfDroppedReplayStrategy;
  import org.apache.activemq.transport.reliable.ReplayBuffer;
  import org.apache.activemq.transport.reliable.ReplayStrategy;
  import org.apache.activemq.transport.reliable.Replayer;
  import org.apache.activemq.util.InetAddressUtil;
  import org.apache.activemq.util.IntSequenceGenerator;
  import org.apache.activemq.util.ServiceStopper;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * An implementation of the {@link Transport} interface using raw UDP
   */
  public class UdpTransport extends TransportThreadSupport implements Transport, Service, Runnable {
  
      private static final Logger LOG = LoggerFactory.getLogger(UdpTransport.class);
  
      private static final int MAX_BIND_ATTEMPTS = 50;
      private static final long BIND_ATTEMPT_DELAY = 100;
  
      private CommandChannel commandChannel;
      private OpenWireFormat wireFormat;
      private ByteBufferPool bufferPool;
      private ReplayStrategy replayStrategy = new ExceptionIfDroppedReplayStrategy();
      private ReplayBuffer replayBuffer;
      private int datagramSize = 4 * 1024;
      private SocketAddress targetAddress;
      private SocketAddress originalTargetAddress;
      private DatagramChannel channel;
      private boolean trace;
      private boolean useLocalHost = false;
      private int port;
      private int minmumWireFormatVersion;
      private String description;
      private IntSequenceGenerator sequenceGenerator;
      private boolean replayEnabled = true;
  
      protected UdpTransport(OpenWireFormat wireFormat) throws IOException {
          this.wireFormat = wireFormat;
      }
  
      public UdpTransport(OpenWireFormat wireFormat, URI remoteLocation) throws UnknownHostException, IOException {
          this(wireFormat);
          this.targetAddress = createAddress(remoteLocation);
          description = remoteLocation.toString() + "@";
      }
  
      public UdpTransport(OpenWireFormat wireFormat, SocketAddress socketAddress) throws IOException {
          this(wireFormat);
          this.targetAddress = socketAddress;
          this.description = getProtocolName() + "ServerConnection@";
      }
  
      /**
       * Used by the server transport
       */
      public UdpTransport(OpenWireFormat wireFormat, int port) throws UnknownHostException, IOException {
          this(wireFormat);
          this.port = port;
          this.targetAddress = null;
          this.description = getProtocolName() + "Server@";
      }
  
      /**
       * Creates a replayer for working with the reliable transport
       */
      public Replayer createReplayer() throws IOException {
          if (replayEnabled) {
              return getCommandChannel();
          }
          return null;
      }
  
      /**
       * A one way asynchronous send
       */
      @Override
      public void oneway(Object command) throws IOException {
          oneway(command, targetAddress);
      }
  
      /**
       * A one way asynchronous send to a given address
       */
      public void oneway(Object command, SocketAddress address) throws IOException {
          if (LOG.isDebugEnabled()) {
              LOG.debug("Sending oneway from: " + this + " to target: " + targetAddress + " command: " + command);
          }
          checkStarted();
          commandChannel.write((Command)command, address);
      }
  
      /**
       * @return pretty print of 'this'
       */
      @Override
      public String toString() {
          if (description != null) {
              return description + port;
          } else {
              return getProtocolUriScheme() + targetAddress + "@" + port;
          }
      }
  
      /**
       * reads packets from a Socket
       */
      @Override
      public void run() {
          LOG.trace("Consumer thread starting for: " + toString());
          while (!isStopped()) {
              try {
                  Command command = commandChannel.read();
                  doConsume(command);
              } catch (AsynchronousCloseException e) {
                  // DatagramChannel closed
                  try {
                      stop();
                  } catch (Exception e2) {
                      LOG.warn("Caught in: " + this + " while closing: " + e2 + ". Now Closed", e2);
                  }
              } catch (SocketException e) {
                  // DatagramSocket closed
                  LOG.debug("Socket closed: " + e, e);
                  try {
                      stop();
                  } catch (Exception e2) {
                      LOG.warn("Caught in: " + this + " while closing: " + e2 + ". Now Closed", e2);
                  }
              } catch (EOFException e) {
                  // DataInputStream closed
                  LOG.debug("Socket closed: " + e, e);
                  try {
                      stop();
                  } catch (Exception e2) {
                      LOG.warn("Caught in: " + this + " while closing: " + e2 + ". Now Closed", e2);
                  }
              } catch (Exception e) {
                  try {
                      stop();
                  } catch (Exception e2) {
                      LOG.warn("Caught in: " + this + " while closing: " + e2 + ". Now Closed", e2);
                  }
                  if (e instanceof IOException) {
                      onException((IOException)e);
                  } else {
                      LOG.error("Caught: " + e, e);
                      e.printStackTrace();
                  }
              }
          }
      }
  
      /**
       * We have received the WireFormatInfo from the server on the actual channel
       * we should use for all future communication with the server, so lets set
       * the target to be the actual channel that the server has chosen for us to
       * talk on.
       */
      public void setTargetEndpoint(Endpoint newTarget) {
          if (newTarget instanceof DatagramEndpoint) {
              DatagramEndpoint endpoint = (DatagramEndpoint)newTarget;
              SocketAddress address = endpoint.getAddress();
              if (address != null) {
                  if (originalTargetAddress == null) {
                      originalTargetAddress = targetAddress;
                  }
                  targetAddress = address;
                  commandChannel.setTargetAddress(address);
              }
          }
      }
  
      // Properties
      // -------------------------------------------------------------------------
      public boolean isTrace() {
          return trace;
      }
  
      public void setTrace(boolean trace) {
          this.trace = trace;
      }
  
      public int getDatagramSize() {
          return datagramSize;
      }
  
      public void setDatagramSize(int datagramSize) {
          this.datagramSize = datagramSize;
      }
  
      public boolean isUseLocalHost() {
          return useLocalHost;
      }
  
      /**
       * Sets whether 'localhost' or the actual local host name should be used to
       * make local connections. On some operating systems such as Macs its not
       * possible to connect as the local host name so localhost is better.
       */
      public void setUseLocalHost(boolean useLocalHost) {
          this.useLocalHost = useLocalHost;
      }
  
      public CommandChannel getCommandChannel() throws IOException {
          if (commandChannel == null) {
              commandChannel = createCommandChannel();
          }
          return commandChannel;
      }
  
      /**
       * Sets the implementation of the command channel to use.
       */
      public void setCommandChannel(CommandDatagramChannel commandChannel) {
          this.commandChannel = commandChannel;
      }
  
      public ReplayStrategy getReplayStrategy() {
          return replayStrategy;
      }
  
      /**
       * Sets the strategy used to replay missed datagrams
       */
      public void setReplayStrategy(ReplayStrategy replayStrategy) {
          this.replayStrategy = replayStrategy;
      }
  
      public int getPort() {
          return port;
      }
  
      /**
       * Sets the port to connect on
       */
      public void setPort(int port) {
          this.port = port;
      }
  
      public int getMinmumWireFormatVersion() {
          return minmumWireFormatVersion;
      }
  
      public void setMinmumWireFormatVersion(int minmumWireFormatVersion) {
          this.minmumWireFormatVersion = minmumWireFormatVersion;
      }
  
      public OpenWireFormat getWireFormat() {
          return wireFormat;
      }
  
      public IntSequenceGenerator getSequenceGenerator() {
          if (sequenceGenerator == null) {
              sequenceGenerator = new IntSequenceGenerator();
          }
          return sequenceGenerator;
      }
  
      public void setSequenceGenerator(IntSequenceGenerator sequenceGenerator) {
          this.sequenceGenerator = sequenceGenerator;
      }
  
      public boolean isReplayEnabled() {
          return replayEnabled;
      }
  
      /**
       * Sets whether or not replay should be enabled when using the reliable
       * transport. i.e. should we maintain a buffer of messages that can be
       * replayed?
       */
      public void setReplayEnabled(boolean replayEnabled) {
          this.replayEnabled = replayEnabled;
      }
  
      public ByteBufferPool getBufferPool() {
          if (bufferPool == null) {
              bufferPool = new DefaultBufferPool();
          }
          return bufferPool;
      }
  
      public void setBufferPool(ByteBufferPool bufferPool) {
          this.bufferPool = bufferPool;
      }
  
      public ReplayBuffer getReplayBuffer() {
          return replayBuffer;
      }
  
      public void setReplayBuffer(ReplayBuffer replayBuffer) throws IOException {
          this.replayBuffer = replayBuffer;
          getCommandChannel().setReplayBuffer(replayBuffer);
      }
  
      // Implementation methods
      // -------------------------------------------------------------------------
  
      /**
       * Creates an address from the given URI
       */
      protected InetSocketAddress createAddress(URI remoteLocation) throws UnknownHostException, IOException {
          String host = resolveHostName(remoteLocation.getHost());
          return new InetSocketAddress(host, remoteLocation.getPort());
      }
  
      protected String resolveHostName(String host) throws UnknownHostException {
          String localName = InetAddressUtil.getLocalHostName();
          if (localName != null && isUseLocalHost()) {
              if (localName.equals(host)) {
                  return "localhost";
              }
          }
          return host;
      }
  
      @Override
      protected void doStart() throws Exception {
          getCommandChannel().start();
  
          super.doStart();
      }
  
      protected CommandChannel createCommandChannel() throws IOException {
          SocketAddress localAddress = createLocalAddress();
          channel = DatagramChannel.open();
  
          channel = connect(channel, targetAddress);
  
          DatagramSocket socket = channel.socket();
          bind(socket, localAddress);
          if (port == 0) {
              port = socket.getLocalPort();
          }
  
          return createCommandDatagramChannel();
      }
  
      protected CommandChannel createCommandDatagramChannel() {
          return new CommandDatagramChannel(this, getWireFormat(), getDatagramSize(), getTargetAddress(), createDatagramHeaderMarshaller(), getChannel(), getBufferPool());
      }
  
      protected void bind(DatagramSocket socket, SocketAddress localAddress) throws IOException {
          channel.configureBlocking(true);
  
          if (LOG.isDebugEnabled()) {
              LOG.debug("Binding to address: " + localAddress);
          }
  
          //
          // We have noticed that on some platfoms like linux, after you close
          // down
          // a previously bound socket, it can take a little while before we can
          // bind it again.
          //
          for (int i = 0; i < MAX_BIND_ATTEMPTS; i++) {
              try {
                  socket.bind(localAddress);
                  return;
              } catch (BindException e) {
                  if (i + 1 == MAX_BIND_ATTEMPTS) {
                      throw e;
                  }
                  try {
                      Thread.sleep(BIND_ATTEMPT_DELAY);
                  } catch (InterruptedException e1) {
                      Thread.currentThread().interrupt();
                      throw e;
                  }
              }
          }
  
      }
  
      protected DatagramChannel connect(DatagramChannel channel, SocketAddress targetAddress2) throws IOException {
          // TODO
          // connect to default target address to avoid security checks each time
          // channel = channel.connect(targetAddress);
  
          return channel;
      }
  
      protected SocketAddress createLocalAddress() {
          return new InetSocketAddress(port);
      }
  
      @Override
      protected void doStop(ServiceStopper stopper) throws Exception {
          if (channel != null) {
              channel.close();
          }
      }
  
      protected DatagramHeaderMarshaller createDatagramHeaderMarshaller() {
          return new DatagramHeaderMarshaller();
      }
  
      protected String getProtocolName() {
          return "Udp";
      }
  
      protected String getProtocolUriScheme() {
          return "udp://";
      }
  
      protected SocketAddress getTargetAddress() {
          return targetAddress;
      }
  
      protected DatagramChannel getChannel() {
          return channel;
      }
  
      protected void setChannel(DatagramChannel channel) {
          this.channel = channel;
      }
  
      public InetSocketAddress getLocalSocketAddress() {
          if (channel == null) {
              return null;
          } else {
              return (InetSocketAddress)channel.socket().getLocalSocketAddress();
          }
      }
  
      @Override
      public String getRemoteAddress() {
          if (targetAddress != null) {
              return "" + targetAddress;
          }
          return null;
      }
  
      @Override
      public int getReceiveCounter() {
          if (commandChannel == null) {
              return 0;
          }
          return commandChannel.getReceiveCounter();
      }
  
      @Override
      public X509Certificate[] getPeerCertificates() {
          return null;
      }
  
      @Override
      public void setPeerCertificates(X509Certificate[] certificates) {
      }
  }
  