1   /**
    2    * Licensed to the Apache Software Foundation (ASF) under one or more
    3    * contributor license agreements.  See the NOTICE file distributed with
    4    * this work for additional information regarding copyright ownership.
    5    * The ASF licenses this file to You under the Apache License, Version 2.0
    6    * (the "License"); you may not use this file except in compliance with
    7    * the License.  You may obtain a copy of the License at
    8    *
    9    *      http://www.apache.org/licenses/LICENSE-2.0
   10    *
   11    * Unless required by applicable law or agreed to in writing, software
   12    * distributed under the License is distributed on an "AS IS" BASIS,
   13    * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   14    * See the License for the specific language governing permissions and
   15    * limitations under the License.
   16    */
   17   package org.apache.activemq.transport.multicast;
   18   
   19   import java.io.IOException;
   20   import java.net.DatagramSocket;
   21   import java.net.InetAddress;
   22   import java.net.InetSocketAddress;
   23   import java.net.MulticastSocket;
   24   import java.net.SocketAddress;
   25   import java.net.SocketException;
   26   import java.net.URI;
   27   import java.net.UnknownHostException;
   28   
   29   import org.apache.activemq.openwire.OpenWireFormat;
   30   import org.apache.activemq.transport.udp.CommandChannel;
   31   import org.apache.activemq.transport.udp.CommandDatagramSocket;
   32   import org.apache.activemq.transport.udp.DatagramHeaderMarshaller;
   33   import org.apache.activemq.transport.udp.UdpTransport;
   34   import org.apache.activemq.util.ServiceStopper;
   35   import org.apache.commons.logging.Log;
   36   import org.apache.commons.logging.LogFactory;
   37   
   38   /**
   39    * A multicast based transport.
   40    * 
   41    * @version $Revision: 564814 $
   42    */
   43   public class MulticastTransport extends UdpTransport {
   44   
   45       private static final Log LOG = LogFactory.getLog(MulticastTransport.class);
   46   
   47       private static final int DEFAULT_IDLE_TIME = 5000;
   48   
   49       private MulticastSocket socket;
   50       private InetAddress mcastAddress;
   51       private int mcastPort;
   52       private int timeToLive = 1;
   53       private boolean loopBackMode;
   54       private long keepAliveInterval = DEFAULT_IDLE_TIME;
   55   
   56       public MulticastTransport(OpenWireFormat wireFormat, URI remoteLocation) throws UnknownHostException, IOException {
   57           super(wireFormat, remoteLocation);
   58       }
   59   
   60       public long getKeepAliveInterval() {
   61           return keepAliveInterval;
   62       }
   63   
   64       public void setKeepAliveInterval(long keepAliveInterval) {
   65           this.keepAliveInterval = keepAliveInterval;
   66       }
   67   
   68       public boolean isLoopBackMode() {
   69           return loopBackMode;
   70       }
   71   
   72       public void setLoopBackMode(boolean loopBackMode) {
   73           this.loopBackMode = loopBackMode;
   74       }
   75   
   76       public int getTimeToLive() {
   77           return timeToLive;
   78       }
   79   
   80       public void setTimeToLive(int timeToLive) {
   81           this.timeToLive = timeToLive;
   82       }
   83   
   84       protected String getProtocolName() {
   85           return "Multicast";
   86       }
   87   
   88       protected String getProtocolUriScheme() {
   89           return "multicast://";
   90       }
   91   
   92       protected void bind(DatagramSocket socket, SocketAddress localAddress) throws SocketException {
   93       }
   94   
   95       protected void doStop(ServiceStopper stopper) throws Exception {
   96           super.doStop(stopper);
   97           if (socket != null) {
   98               try {
   99                   socket.leaveGroup(getMulticastAddress());
  100               } catch (IOException e) {
  101                   stopper.onException(this, e);
  102               }
  103               socket.close();
  104           }
  105       }
  106   
  107       protected CommandChannel createCommandChannel() throws IOException {
  108           socket = new MulticastSocket(mcastPort);
  109           socket.setLoopbackMode(loopBackMode);
  110           socket.setTimeToLive(timeToLive);
  111   
  112           LOG.debug("Joining multicast address: " + getMulticastAddress());
  113           socket.joinGroup(getMulticastAddress());
  114           socket.setSoTimeout((int)keepAliveInterval);
  115   
  116           return new CommandDatagramSocket(this, getWireFormat(), getDatagramSize(), getTargetAddress(), createDatagramHeaderMarshaller(), getSocket());
  117       }
  118   
  119       protected InetAddress getMulticastAddress() {
  120           return mcastAddress;
  121       }
  122   
  123       protected MulticastSocket getSocket() {
  124           return socket;
  125       }
  126   
  127       protected void setSocket(MulticastSocket socket) {
  128           this.socket = socket;
  129       }
  130   
  131       protected InetSocketAddress createAddress(URI remoteLocation) throws UnknownHostException, IOException {
  132           this.mcastAddress = InetAddress.getByName(remoteLocation.getHost());
  133           this.mcastPort = remoteLocation.getPort();
  134           return new InetSocketAddress(mcastAddress, mcastPort);
  135       }
  136   
  137       protected DatagramHeaderMarshaller createDatagramHeaderMarshaller() {
  138           return new MulticastDatagramHeaderMarshaller("udp://dummyHostName:" + getPort());
  139       }
  140   
  141   }