1 /*

     2  * Copyright 2007-2008 Sun Microsystems, Inc.  All Rights Reserved.

     3  *

     4  * Redistribution and use in source and binary forms, with or without

     5  * modification, are permitted provided that the following conditions

     6  * are met:

     7  *

     8  *   - Redistributions of source code must retain the above copyright

     9  *     notice, this list of conditions and the following disclaimer.

    10  *

    11  *   - Redistributions in binary form must reproduce the above copyright

    12  *     notice, this list of conditions and the following disclaimer in the

    13  *     documentation and/or other materials provided with the distribution.

    14  *

    15  *   - Neither the name of Sun Microsystems nor the names of its

    16  *     contributors may be used to endorse or promote products derived

    17  *     from this software without specific prior written permission.

    18  *

    19  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS

    20  * IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,

    21  * THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR

    22  * PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR

    23  * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,

    24  * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,

    25  * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR

    26  * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF

    27  * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING

    28  * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS

    29  * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

    30  */

    31 

    32 import java.nio.channels.*;

    33 import java.nio.charset.*;

    34 import java.nio.ByteBuffer;

    35 import java.net.*;

    36 import java.io.IOException;

    37 import java.util.*;

    38 

    39 public class Reader {

    40 

    41     static void usage() {

    42         System.err.println("usage: java Reader group:port@interf [-only source...] [-block source...]");

    43         System.exit(-1);

    44     }

    45 

    46     static void printDatagram(SocketAddress sa, ByteBuffer buf) {

    47         System.out.format("-- datagram from %s --\n",

    48             ((InetSocketAddress)sa).getAddress().getHostAddress());

    49         System.out.println(Charset.defaultCharset().decode(buf));

    50     }

    51 

    52     static void parseAddessList(String s, List<InetAddress> list)

    53         throws UnknownHostException

    54     {

    55         String[] sources = s.split(",");

    56         for (int i=0; i<sources.length; i++) {

    57             list.add(InetAddress.getByName(sources[i]));

    58         }

    59     }

    60 

    61     public static void main(String[] args) throws IOException {

    62         if (args.length == 0)

    63             usage();

    64 

    65         // first parameter is the multicast address (interface required)

    66         MulticastAddress target = MulticastAddress.parse(args[0]);

    67         if (target.interf() == null)

    68             usage();

    69 

    70         // addition arguments are source addresses to include or exclude

    71         List<InetAddress> includeList = new ArrayList<InetAddress>();

    72         List<InetAddress> excludeList = new ArrayList<InetAddress>();

    73         int argc = 1;

    74         while (argc < args.length) {

    75             String option = args[argc++];

    76             if (argc >= args.length)

    77                 usage();

    78             String value = args[argc++];

    79             if (option.equals("-only")) {

    80                  parseAddessList(value, includeList);

    81                 continue;

    82             }

    83             if (option.equals("-block")) {

    84                 parseAddessList(value, excludeList);

    85                 continue;

    86             }

    87             usage();

    88         }

    89         if (!includeList.isEmpty() && !excludeList.isEmpty()) {

    90             usage();

    91         }

    92 

    93         // create and bind socket

    94         ProtocolFamily family = StandardProtocolFamily.INET;

    95         if (target.group() instanceof Inet6Address) {

    96             family = StandardProtocolFamily.INET6;

    97         }

    98         DatagramChannel dc = DatagramChannel.open(family)

    99             .setOption(StandardSocketOption.SO_REUSEADDR, true)

   100             .bind(new InetSocketAddress(target.port()));

   101 

   102         if (includeList.isEmpty()) {

   103             // join group and block addresses on the exclude list

   104             MembershipKey key = dc.join(target.group(), target.interf());

   105             for (InetAddress source: excludeList) {

   106                 key.block(source);

   107             }

   108         } else {

   109             // join with source-specific membership for each source

   110             for (InetAddress source: includeList) {

   111                 dc.join(target.group(), target.interf(), source);

   112             }

   113         }

   114 

   115         // register socket with Selector

   116         Selector sel = Selector.open();

   117         dc.configureBlocking(false);

   118         dc.register(sel, SelectionKey.OP_READ);

   119 

   120         // print out each datagram that we receive

   121         ByteBuffer buf = ByteBuffer.allocateDirect(4096);

   122         for (;;) {

   123             int updated = sel.select();

   124             if (updated > 0) {

   125                 Iterator<SelectionKey> iter = sel.selectedKeys().iterator();

   126                 while (iter.hasNext()) {

   127                     SelectionKey sk = iter.next();

   128                     iter.remove();

   129 

   130                     DatagramChannel ch = (DatagramChannel)sk.channel();

   131                     SocketAddress sa = ch.receive(buf);

   132                     if (sa != null) {

   133                         buf.flip();

   134                         printDatagram(sa, buf);

   135                         buf.rewind();

   136                         buf.limit(buf.capacity());

   137                     }

   138                 }

   139             }

   140         }

   141     }

   142 }