import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
/**
 * Constructs new instance.
 *
 * @param cfg Configuration.
 *
 * @throws UnknownHostException If failed to resolve multicast group address.
 */
public MulticastSeedNodeProvider(MulticastSeedNodeProviderConfig cfg) throws UnknownHostException {
    ConfigCheck check = ConfigCheck.get(getClass());

    check.notNull(cfg, "configuration");
    check.positive(cfg.getPort(), "port");
    check.nonNegative(cfg.getTtl(), "TTL");
    check.notEmpty(cfg.getGroup(), "multicast group");
    check.positive(cfg.getInterval(), "discovery interval");
    check.positive(cfg.getWaitTime(), "wait time");
    check.that(cfg.getInterval() < cfg.getWaitTime(), "discovery interval must be greater than wait time "
        + "[discovery-interval=" + cfg.getInterval() + ", wait-time=" + cfg.getWaitTime() + ']');

    InetAddress groupAddress = InetAddress.getByName(cfg.getGroup());

    check.isTrue(groupAddress.isMulticastAddress(), "address is not a multicast address [address=" + groupAddress + ']');

    group = new InetSocketAddress(groupAddress, cfg.getPort());
    ttl = cfg.getTtl();
    interval = cfg.getInterval();
    waitTime = cfg.getWaitTime();
    loopBackDisabled = cfg.isLoopBackDisabled();

    ipVer = group.getAddress() instanceof Inet6Address ? InternetProtocolFamily.IPv6 : InternetProtocolFamily.IPv4;
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
@Override
public void subscribe(final Subscriber<? super O> subscriber) {
  try {
    final List<NetworkInterface> interfaces = Collections
        .list(NetworkInterface.getNetworkInterfaces());

    UdpServer
        .create(opts ->
            opts.option(ChannelOption.SO_REUSEADDR, true)
                .connectAddress(() -> new InetSocketAddress(this.port))
                .protocolFamily(InternetProtocolFamily.IPv4))
        .newHandler((in, out) -> {
          Flux.fromIterable(interfaces)
              .flatMap(iface -> in.join(this.address, iface))
              .thenMany(in.receive().asByteArray())
              .map(this.parser)
              .subscribe(subscriber);
          return Flux.never();
        })
        .subscribe();
  } catch (final SocketException exception) {
    Flux.<O>error(exception).subscribe(subscriber);
  }
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
public MulticastServerHandler(Cluster cluster, InetSocketAddress address) throws InterruptedException {
    this.address = address;

    handler = new Bootstrap()
            .channelFactory(() -> new NioDatagramChannel(InternetProtocolFamily.IPv4))
            .localAddress(address)
            .group(new NioEventLoopGroup())
            .option(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.IP_MULTICAST_IF, multicastInterface)
            .option(ChannelOption.AUTO_READ, false)
            .handler(new ChannelInitializer<NioDatagramChannel>() {
                @Override
                public void initChannel(NioDatagramChannel ch) throws Exception {
                    ch.pipeline().addLast(new MulticastChannelAdapter(cluster));
                }
            });
    localMember = cluster.getLocalMember();
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
@Override
public void subscribe(final Subscriber<? super DatagramPacket> subscriber) {
  UdpClient
      .create(opts -> opts
          .option(ChannelOption.SO_REUSEADDR, true)
          .port(this.port)
          .protocolFamily(InternetProtocolFamily.IPv4))
      .newHandler((in, out) -> {
        in.receive().asByteArray()
            .map(bytes -> new DatagramPacket(bytes, bytes.length))
            .subscribe(subscriber);
        return Flux.never();
      })
      .subscribe();
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
@Override
public T newChannel() {
    try {
      return ReflectionUtils.instantiateWithCustomCtor(clazz.getName(),
        new Class[] { InternetProtocolFamily.class }, new Object[] { family });

    } catch (Throwable t) {
        throw new ChannelException("Unable to create Channel from class " + clazz, t);
    }
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
public List<BootstrapComboFactory<Bootstrap, Bootstrap>> datagram() {
    // Make the list of Bootstrap factories.
    List<BootstrapFactory<Bootstrap>> bfs = Arrays.asList(
            new BootstrapFactory<Bootstrap>() {
                @Override
                public Bootstrap newInstance() {
                    return new Bootstrap().group(nioWorkerGroup).channelFactory(new ChannelFactory<Channel>() {
                        @Override
                        public Channel newChannel() {
                            return new NioDatagramChannel(InternetProtocolFamily.IPv4);
                        }

                        @Override
                        public String toString() {
                            return NioDatagramChannel.class.getSimpleName() + ".class";
                        }
                    });
                }
            },
            new BootstrapFactory<Bootstrap>() {
                @Override
                public Bootstrap newInstance() {
                    return new Bootstrap().group(oioWorkerGroup).channel(OioDatagramChannel.class);
                }
            }
    );

    // Populare the combinations.
    return combo(bfs, bfs);
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
/**
 * Convert the {@link InternetProtocolFamily}. This MUST only be called on jdk version >= 7.
 */
public static ProtocolFamily convert(InternetProtocolFamily family) {
    switch (family) {
    case IPv4:
        return StandardProtocolFamily.INET;
    case IPv6:
        return StandardProtocolFamily.INET6;
    default:
        throw new IllegalArgumentException();
    }
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
private static DatagramChannel newSocket(SelectorProvider provider, InternetProtocolFamily ipFamily) {
    if (ipFamily == null) {
        return newSocket(provider);
    }

    checkJavaVersion();

    try {
        return provider.openDatagramChannel(ProtocolFamilyConverter.convert(ipFamily));
    } catch (IOException e) {
        throw new ChannelException("Failed to open a socket.", e);
    }
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
@Override
public List<TestsuitePermutation.BootstrapComboFactory<Bootstrap, Bootstrap>> datagram() {
    // Make the list of Bootstrap factories.
    List<BootstrapFactory<Bootstrap>> bfs = Arrays.asList(
            new BootstrapFactory<Bootstrap>() {
                @Override
                public Bootstrap newInstance() {
                    return new Bootstrap().group(nioWorkerGroup).channelFactory(new ChannelFactory<Channel>() {
                        @Override
                        public Channel newChannel() {
                            return new NioDatagramChannel(InternetProtocolFamily.IPv4);
                        }

                        @Override
                        public String toString() {
                            return NioDatagramChannel.class.getSimpleName() + ".class";
                        }
                    });
                }
            },
            new BootstrapFactory<Bootstrap>() {
                @Override
                public Bootstrap newInstance() {
                    return new Bootstrap().group(EPOLL_WORKER_GROUP).channel(EpollDatagramChannel.class);
                }
            }
    );
    return combo(bfs, bfs);
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
private static DatagramChannel newSocket(InternetProtocolFamily ipFamily) {
    if (ipFamily == null) {
        return newSocket();
    }

    if (PlatformDependent.javaVersion() < 7) {
        throw new UnsupportedOperationException();
    }

    try {
        return DatagramChannel.open(ProtocolFamilyConverter.convert(ipFamily));
    } catch (IOException e) {
        throw new ChannelException("Failed to open a socket.", e);
    }
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
public DataServerThread() {
    dataBossGroup = new NioEventLoopGroup(DATA_BOSS_THREADS);
    dataWorkerGroup = new NioEventLoopGroup(DATA_WORKER_THREADS);
    try {
        Bootstrap b = new Bootstrap();
        b.group(dataWorkerGroup)
                .channelFactory(new ChannelFactory<Channel>() {
                    @Override
                    public Channel newChannel() {
                        return new NioDatagramChannel(InternetProtocolFamily.IPv4);
                    }

                    @Override
                    public String toString() {
                        return NioDatagramChannel.class.getSimpleName() + ".class";
                    }
                }).handler(new ChannelInitializer<DatagramChannel>() {
                    @Override
                    public void initChannel(DatagramChannel ch) throws Exception {
                        ch.config().setAllocator(UnpooledByteBufAllocator.DEFAULT);
                        ch.pipeline().addLast(new DataMessageHandler());
                        if (LOG.isTraceEnabled()) {
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.TRACE));
                        }
                    }

                    @Override
                    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                        LOG.error("Cannot initialize data server.", cause);
                    }
                });

        // Bind and start to accept incoming connections.
        f = b.bind(getIp(), getDataPort()).sync();
    } catch (InterruptedException ex) {
        LOG.error("Message data interrupted.", ex);
    }
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
HBaseDatagramChannelFactory(Class<? extends T> clazz, InternetProtocolFamily family) {
    this.clazz = clazz;
    this.family = family;
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
public NioDatagramChannelFactory(InternetProtocolFamily ipFamily) {
    this.ipFamily = ipFamily;
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
static List<Entry<Factory<Bootstrap>, Factory<Bootstrap>>> datagram() {
    List<Entry<Factory<Bootstrap>, Factory<Bootstrap>>> list =
            new ArrayList<Entry<Factory<Bootstrap>, Factory<Bootstrap>>>();

    // Make the list of Bootstrap factories.
    List<Factory<Bootstrap>> bfs =
            new ArrayList<Factory<Bootstrap>>();
    bfs.add(new Factory<Bootstrap>() {
        @Override
        public Bootstrap newInstance() {
            return new Bootstrap().group(nioWorkerGroup).channelFactory(new ChannelFactory<Channel>() {
                @Override
                public Channel newChannel() {
                   return new NioDatagramChannel(InternetProtocolFamily.IPv4);
                }

                @Override
                public String toString() {
                    return NioDatagramChannel.class.getSimpleName() + ".class";
                }
            });
        }
    });
    bfs.add(new Factory<Bootstrap>() {
        @Override
        public Bootstrap newInstance() {
            return new Bootstrap().group(oioWorkerGroup).channel(OioDatagramChannel.class);
        }
    });

    // Populate the combinations
    for (Factory<Bootstrap> sbf: bfs) {
        for (Factory<Bootstrap> cbf: bfs) {
            final Factory<Bootstrap> sbf0 = sbf;
            final Factory<Bootstrap> cbf0 = cbf;
            list.add(new Entry<Factory<Bootstrap>, Factory<Bootstrap>>() {
                @Override
                public Factory<Bootstrap> getKey() {
                    return sbf0;
                }

                @Override
                public Factory<Bootstrap> getValue() {
                    return cbf0;
                }

                @Override
                public Factory<Bootstrap> setValue(Factory<Bootstrap> value) {
                    throw new UnsupportedOperationException();
                }
            });
        }
    }

    return list;
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
static List<Entry<Factory<Bootstrap>, Factory<Bootstrap>>> datagram() {
    List<Entry<Factory<Bootstrap>, Factory<Bootstrap>>> list =
            new ArrayList<Entry<Factory<Bootstrap>, Factory<Bootstrap>>>();

    // Make the list of Bootstrap factories.
    List<Factory<Bootstrap>> bfs = new ArrayList<Factory<Bootstrap>>();
    bfs.add(new Factory<Bootstrap>() {
        @Override
        public Bootstrap newInstance() {
            return new Bootstrap().group(nioWorkerGroup).channelFactory(new ChannelFactory<Channel>() {
                @Override
                public Channel newChannel(EventLoop eventLoop) {
                   return new NioDatagramChannel(eventLoop, InternetProtocolFamily.IPv4);
                }

                @Override
                public String toString() {
                    return NioDatagramChannel.class.getSimpleName() + ".class";
                }
            });
        }
    });
    bfs.add(new Factory<Bootstrap>() {
        @Override
        public Bootstrap newInstance() {
            return new Bootstrap().group(oioWorkerGroup).channel(OioDatagramChannel.class);
        }
    });

    // Populate the combinations
    for (Factory<Bootstrap> sbf: bfs) {
        for (Factory<Bootstrap> cbf: bfs) {
            final Factory<Bootstrap> sbf0 = sbf;
            final Factory<Bootstrap> cbf0 = cbf;
            list.add(new Entry<Factory<Bootstrap>, Factory<Bootstrap>>() {
                @Override
                public Factory<Bootstrap> getKey() {
                    return sbf0;
                }

                @Override
                public Factory<Bootstrap> getValue() {
                    return cbf0;
                }

                @Override
                public Factory<Bootstrap> setValue(Factory<Bootstrap> value) {
                    throw new UnsupportedOperationException();
                }
            });
        }
    }

    return list;
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
/**
 * Create a new instance using the given {@link InternetProtocolFamily}. If {@code null} is used it will depend
 * on the Operation Systems default which will be chosen.
 */
public NioDatagramChannel(InternetProtocolFamily ipFamily) {
    this(newSocket(DEFAULT_SELECTOR_PROVIDER, ipFamily));
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
/**
 * Create a new instance using the given {@link InternetProtocolFamily}. If {@code null} is used it will depend
 * on the Operation Systems default which will be chosen.
 */
public NioDatagramChannel(InternetProtocolFamily ipFamily) {
    this(newSocket(ipFamily));
}
 

import io.netty.channel.socket.InternetProtocolFamily; //导入依赖的package包/类
/**
 * Create a new instance using the given {@link InternetProtocolFamily}. If {@code null} is used it will depend
 * on the Operation Systems default which will be chosen.
 */
public NioDatagramChannel(EventLoop eventLoop, InternetProtocolFamily ipFamily) {
    this(eventLoop, newSocket(ipFamily));
}