private static void doConnect(
            final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise connectPromise) {

        // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
        // the pipeline in its channelRegistered() implementation.//这个方法在触发channelRegistered()之前被调用。给用户处理程序一个设置的机会
//管道的channelRegistered()实现。
        final Channel channel = connectPromise.channel();
        channel.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                if (localAddress == null) {
//                    远程连接
                    channel.connect(remoteAddress, connectPromise);
                } else {
//                    本地连接
                    channel.connect(remoteAddress, localAddress, connectPromise);
                }
                connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        });
    }
 

void runWithManyOffsets(SocketAddress addr, int bufferSize)
    throws IOException
{
    doTest(addr, bufferSize, 1);
    doTest(addr, bufferSize, 2);
    doTest(addr, bufferSize, 3);
    doTest(addr, bufferSize, 4);
    doTest(addr, bufferSize, 5);
    doTest(addr, bufferSize, 6);
    doTest(addr, bufferSize, 7);
    doTest(addr, bufferSize, 8);
    doTest(addr, bufferSize, 9);
    doTest(addr, bufferSize, 10);
    doTest(addr, bufferSize, 11);
    doTest(addr, bufferSize, 12);
    doTest(addr, bufferSize, 13);
    doTest(addr, bufferSize, 14);
    doTest(addr, bufferSize, 15);
}
 

@Test
public void create() {
  List<EquivalentAddressGroup> addrs = new ArrayList<>();
  addrs.add(new EquivalentAddressGroup(mock(SocketAddress.class)));
  Attributes attr = Attributes.newBuilder().build();
  ChildLbResolvedAddressFactory factory = new ChildLbResolvedAddressFactory(addrs, attr);
  Object config1 = new Object();
  
  ResolvedAddresses resolvedAddress = factory.create(config1);
  
  assertThat(resolvedAddress.getAddresses()).isEqualTo(addrs);
  assertThat(resolvedAddress.getAttributes()).isEqualTo(attr);
  assertThat(resolvedAddress.getLoadBalancingPolicyConfig()).isEqualTo(config1);

  Object config2 = "different object";
  
  resolvedAddress = factory.create(config2);

  assertThat(resolvedAddress.getAddresses()).isEqualTo(addrs);
  assertThat(resolvedAddress.getAttributes()).isEqualTo(attr);
  assertThat(resolvedAddress.getLoadBalancingPolicyConfig()).isEqualTo(config2);
}
 

@Test
public void socketChannelShouldAlwaysReturnTheSameSocket() throws Exception {
	final SocketAddress socketAddress = new InetSocketAddress(this.hostname, this.port);
	final int timeout = 1;

	final SocketChannel socketChannel = SocketChannel.open();

	final Socket socket = socketChannel.socket();
	try {
		socket.connect(socketAddress, timeout);
	} catch (SocketTimeoutException | ConnectException e) { // NOPMD (empty catch block)
		// both of the exceptions indicate a connection timeout
		// => ignore to reconnect
	}

	// The previous connect should throw an exception
	assertThat(socket.isConnected(), is(false)); // NOPMD (JUnit message is not necessary)
}
 

@Override
public final AsynchronousSocketChannel bind(SocketAddress local)
    throws IOException
{
    try {
        begin();
        synchronized (stateLock) {
            if (state == ST_PENDING)
                throw new ConnectionPendingException();
            if (localAddress != null)
                throw new AlreadyBoundException();
            InetSocketAddress isa = (local == null) ?
                new InetSocketAddress(0) : Net.checkAddress(local);
            SecurityManager sm = System.getSecurityManager();
            if (sm != null) {
                sm.checkListen(isa.getPort());
            }
            NetHooks.beforeTcpBind(fd, isa.getAddress(), isa.getPort());
            Net.bind(fd, isa.getAddress(), isa.getPort());
            localAddress = Net.localAddress(fd);
        }
    } finally {
        end();
    }
    return this;
}
 

/**
 * This utility method tries to connect to the JobManager using the InetAddress returned by
 * InetAddress.getLocalHost(). The purpose of the utility is to have a final try connecting to
 * the target address using the LocalHost before using the address returned.
 * We do a second try because the JM might have been unavailable during the first check.
 *
 * @param preliminaryResult The address detected by the heuristic
 * @return either the preliminaryResult or the address returned by InetAddress.getLocalHost() (if
 * 			we are able to connect to targetAddress from there)
 */
private static InetAddress tryLocalHostBeforeReturning(
			InetAddress preliminaryResult, SocketAddress targetAddress, boolean logging) throws IOException {

	InetAddress localhostName = InetAddress.getLocalHost();

	if (preliminaryResult.equals(localhostName)) {
		// preliminary result is equal to the local host name
		return preliminaryResult;
	}
	else if (tryToConnect(localhostName, targetAddress, AddressDetectionState.SLOW_CONNECT.getTimeout(), logging)) {
		// success, we were able to use local host to connect
		LOG.debug("Preferring {} (InetAddress.getLocalHost()) for local bind point over previous candidate {}",
				localhostName, preliminaryResult);
		return localhostName;
	}
	else {
		// we have to make the preliminary result the final result
		return preliminaryResult;
	}
}
 

@Override
public final AsynchronousServerSocketChannel bind(SocketAddress local, int backlog)
    throws IOException
{
    InetSocketAddress isa = (local == null) ? new InetSocketAddress(0) :
        Net.checkAddress(local);
    SecurityManager sm = System.getSecurityManager();
    if (sm != null)
        sm.checkListen(isa.getPort());

    try {
        begin();
        synchronized (stateLock) {
            if (localAddress != null)
                throw new AlreadyBoundException();
            NetHooks.beforeTcpBind(fd, isa.getAddress(), isa.getPort());
            Net.bind(fd, isa.getAddress(), isa.getPort());
            Net.listen(fd, backlog < 1 ? 50 : backlog);
            localAddress = Net.localAddress(fd);
        }
    } finally {
        end();
    }
    return this;
}
 

@Override
public Set<SocketAddress> getRemoteAddresses()
        throws IOException {
    synchronized (stateLock) {
        if (!isOpen())
            throw new ClosedChannelException();
        if (!isConnected() || isShutdown)
            return Collections.emptySet();

        try {
            return SctpNet.getRemoteAddresses(fdVal, 0/*unused*/);
        } catch (SocketException unused) {
            /* an open connected channel should always have remote addresses */
            return remoteAddresses;
        }
    }
}
 

@Override
public boolean connect(SocketAddress remote) throws IOException {

    long start = System.currentTimeMillis();
    //log.debug("trying to connect");
    socketChannel.connect(remote);
    if (selector.select(connectTimeout) > 0) {
        selector.selectedKeys().remove(channelKey);
        //log.debug("selected connect");
        //log.debug("Spent " + (System.currentTimeMillis() - start));
        if (!channelKey.isConnectable()) {
            throw new IllegalStateException("Socket channel is in not connectable state");
        }

        socketChannel.finishConnect();
        channelKey = socketChannel.register(selector, SelectionKey.OP_READ);
        if (log.isDebugEnabled()) {
            log.debug("Connected socket in " + (System.currentTimeMillis() - start));
        }
        if (!socketChannel.isConnected()) {
            throw new SocketException("SocketChannel not connected on some reason");
        }
        return true;
    }
    //log.debug("Spent " + (System.currentTimeMillis() - start));
    throw new SocketTimeoutException("Failed to connect to " + remote.toString());
}
 

@Override
public Set<SocketAddress> getAllLocalAddresses()
        throws IOException {
    synchronized (stateLock) {
        if (!isOpen())
            throw new ClosedChannelException();
        if (!isBound())
            return Collections.emptySet();

        return SctpNet.getLocalAddresses(fdVal);
    }
}
 

@Test
public void subchannelChannel_normalUsage() {
  createChannel();
  Subchannel subchannel = createSubchannelSafely(helper, addressGroup, Attributes.EMPTY);
  verify(balancerRpcExecutorPool, never()).getObject();

  Channel sChannel = subchannel.asChannel();
  verify(balancerRpcExecutorPool).getObject();

  Metadata headers = new Metadata();
  CallOptions callOptions = CallOptions.DEFAULT.withDeadlineAfter(5, TimeUnit.SECONDS);

  // Subchannel must be READY when creating the RPC.
  subchannel.requestConnection();
  verify(mockTransportFactory)
      .newClientTransport(any(SocketAddress.class), any(ClientTransportOptions.class));
  MockClientTransportInfo transportInfo = transports.poll();
  ConnectionClientTransport mockTransport = transportInfo.transport;
  ManagedClientTransport.Listener transportListener = transportInfo.listener;
  transportListener.transportReady();

  ClientCall<String, Integer> call = sChannel.newCall(method, callOptions);
  call.start(mockCallListener, headers);
  verify(mockTransport).newStream(same(method), same(headers), callOptionsCaptor.capture());

  CallOptions capturedCallOption = callOptionsCaptor.getValue();
  assertThat(capturedCallOption.getDeadline()).isSameAs(callOptions.getDeadline());
  assertThat(capturedCallOption.getOption(GrpcUtil.CALL_OPTIONS_RPC_OWNED_BY_BALANCER)).isTrue();
}
 

@Override
public ListenableFuture<SocketAddress> openConnection(SocketAddress serverAddress, StreamConnection connection) {
    try {
        if (!isRunning())
            throw new IllegalStateException();
        return new BlockingClient(serverAddress, connection, connectTimeoutMillis, socketFactory, clients).getConnectFuture();
    } catch (IOException e) {
        throw new RuntimeException(e); // This should only happen if we are, eg, out of system resources
    }
}
 

public Peer findOrNewPeer(@NotNull SocketAddress address) {
    Peer peer = peers.get(address);
    if (peer == null) {
        peers.putIfAbsent(address, new Peer(this, address));
        peer = peers.get(address);
    }
    return peer;
}
 

/**
 * Select one of the configured local interfaces at random. We use a random
 * interface because other policies like round-robin are less effective
 * given that we cache connections to datanodes.
 *
 * @return one of the local interface addresses at random, or null if no
 *    local interfaces are configured
 */
SocketAddress getRandomLocalInterfaceAddr() {
  if (localInterfaceAddrs.length == 0) {
    return null;
  }
  final int idx = r.nextInt(localInterfaceAddrs.length);
  final SocketAddress addr = localInterfaceAddrs[idx];
  if (LOG.isDebugEnabled()) {
    LOG.debug("Using local interface " + addr);
  }
  return addr;
}
 

public Server() throws IOException {
    ssc = SctpServerChannel.open().bind(null);
    java.util.Set<SocketAddress> addrs = ssc.getAllLocalAddresses();
    if (addrs.isEmpty())
        debug("addrs should not be empty");

    serverAddr = (InetSocketAddress) addrs.iterator().next();
}
 

@Override
public void bind(
        ChannelHandlerContext ctx,
        SocketAddress localAddress, ChannelPromise promise) throws Exception {
    assert ctx == outboundCtx.ctx;
    if (!outboundCtx.removed) {
        outboundHandler.bind(outboundCtx, localAddress, promise);
    } else {
        outboundCtx.bind(localAddress, promise);
    }
}
 

private void connectInternal(SocketAddress remote) throws SocketException {
    try {
        dc.connect(remote, false); // skips check for already connected
    } catch (ClosedChannelException e) {
        // ignore
    } catch (Exception x) {
        Net.translateToSocketException(x);
    }
}
 

public static String parseSocketAddressName(SocketAddress socketAddress) {

        final InetSocketAddress addrs = (InetSocketAddress) socketAddress;
        if (addrs != null) {
            return addrs.getAddress().getHostName();
        }
        return "";
    }
 

@Override
protected SocketAddress localAddress0() {
    StreamConnection conn = connection();
    if (conn == null) {
        return null;
    }
    return conn.getLocalAddress();
}
 

/**
 * {@inheritDoc}
 */
public final void bind(SocketAddress... addresses) throws IOException {
    if ((addresses == null) || (addresses.length == 0)) {
        bind(getDefaultLocalAddresses());
        return;
    }

    List<SocketAddress> localAddresses = new ArrayList<SocketAddress>(2);

    for (SocketAddress address : addresses) {
        localAddresses.add(address);
    }

    bind(localAddresses);
}
 

@Override
    public void connect() throws IOException {

        L.d("RtmpConnection.connect() called. Host: " + host + ", port: " + port + ", appName: " + appName + ", playPath: " + streamName);
        socket = new Socket();
        SocketAddress socketAddress = new InetSocketAddress(host, port);
        socket.connect(socketAddress, SOCKET_CONNECT_TIMEOUT_MS);
        BufferedInputStream in = new BufferedInputStream(socket.getInputStream());
        BufferedOutputStream out = new BufferedOutputStream(socket.getOutputStream());
        L.d("RtmpConnection.connect(): socket connection established, doing handhake...");
        handshake(in, out);
        active = true;
        L.d("RtmpConnection.connect(): handshake done");
        readThread = new ReadThread(rtmpSessionInfo, in, this, this);
        writeThread = new WriteThread(rtmpSessionInfo, out, this);
        readThread.start();
        writeThread.start();

        // Start the "main" handling thread
        new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    L.d("RtmpConnection: starting main rx handler loop");
                    handleRxPacketLoop();
                } catch (IOException ex) {
                    Logger.getLogger(RtmpConnection.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }).start();
//        rxHandler;

        rtmpConnect();
    }
 

public void sessionClosed(Object session, SyslogServerIF syslogServer, SocketAddress socketAddress, boolean timeout) {
	if (session != null) {
		int i = translate((String) session);
		
		if (i != -1) {
			closeCount[i]++;
		}
	}

	System.out.println("closed: " + id + "/" + session.toString());
}
 

@Override
public boolean connect(SocketAddress endpoint,
                       int maxOutStreams,
                       int maxInStreams)
        throws IOException {
    ensureOpenAndUnconnected();
    return setOption(SCTP_INIT_MAXSTREAMS, InitMaxStreams.
            create(maxInStreams, maxOutStreams)).connect(endpoint);

}
 

private Map<String, String> extractMessageAttributes() {
    final Map<String, String> attributes = new HashMap<>();
    final Certificate[] tlsPeerCertificates = context.getTlsPeerCertificates();
    if (tlsPeerCertificates != null) {
        for (int i = 0; i < tlsPeerCertificates.length; i++) {
            if (tlsPeerCertificates[i] instanceof X509Certificate) {
                X509Certificate x509Cert = (X509Certificate) tlsPeerCertificates[i];
                attributes.put("smtp.certificate." + i + ".serial", x509Cert.getSerialNumber().toString());
                attributes.put("smtp.certificate." + i + ".subjectName", x509Cert.getSubjectDN().getName());
            }
        }
    }

    SocketAddress address = context.getRemoteAddress();
    if (address != null) {
        // will extract and format source address if available
        String strAddress = address instanceof InetSocketAddress
                ? ((InetSocketAddress) address).getHostString() + ":" + ((InetSocketAddress) address).getPort()
                : context.getRemoteAddress().toString();
        attributes.put("smtp.src", strAddress);
    }

    attributes.put("smtp.helo", context.getHelo());
    attributes.put("smtp.from", from);
    for (int i = 0; i < recipientList.size(); i++) {
        attributes.put("smtp.recipient." + i, recipientList.get(i));
    }
    attributes.put(CoreAttributes.MIME_TYPE.key(), "message/rfc822");
    return attributes;
}
 

private TNonblockingSocket(SocketChannel socketChannel, int timeout, SocketAddress socketAddress)
    throws IOException {
  socketChannel_ = socketChannel;
  socketAddress_ = socketAddress;

  // make it a nonblocking channel
  socketChannel.configureBlocking(false);

  // set options
  Socket socket = socketChannel.socket();
  socket.setSoLinger(false, 0);
  socket.setTcpNoDelay(true);
  setTimeout(timeout);
}
 

public Server() throws IOException {
    ssc = SctpServerChannel.open().bind(null);
    java.util.Set<SocketAddress> addrs = ssc.getAllLocalAddresses();
    if (addrs.isEmpty())
        debug("addrs should not be empty");

    serverAddr = (InetSocketAddress) addrs.iterator().next();
}
 

private static CompatClient serviceTalkClient(final SocketAddress serverAddress, final boolean ssl) {
    final GrpcClientBuilder<InetSocketAddress, InetSocketAddress> builder =
            GrpcClients.forResolvedAddress((InetSocketAddress) serverAddress);
    if (ssl) {
        builder.secure().disableHostnameVerification().provider(OPENSSL)
                .trustManager(DefaultTestCerts::loadServerPem).commit();
    }
    return builder.build(new Compat.ClientFactory());
}
 

private static Proxy getProxy ( final String url )
{
    final ProxySelector ps = ProxySelector.getDefault ();
    if ( ps == null )
    {
        logger.debug ( "No proxy selector found" );
        return null;
    }

    final List<java.net.Proxy> proxies = ps.select ( URI.create ( url ) );
    for ( final java.net.Proxy proxy : proxies )
    {
        if ( proxy.type () != Type.HTTP )
        {
            logger.debug ( "Unsupported proxy type: {}", proxy.type () );
            continue;
        }

        final SocketAddress addr = proxy.address ();
        logger.debug ( "Proxy address: {}", addr );

        if ( ! ( addr instanceof InetSocketAddress ) )
        {
            logger.debug ( "Unsupported proxy address type: {}", addr.getClass () );
            continue;
        }

        final InetSocketAddress inetAddr = (InetSocketAddress)addr;

        return new Proxy ( Proxy.TYPE_HTTP, inetAddr.getHostString (), inetAddr.getPort () );
    }

    logger.debug ( "No proxy found" );
    return null;
}
 

protected static void sendRejectedReply(RpcCall call,
    SocketAddress remoteAddress, ChannelHandlerContext ctx) {
  XDR out = new XDR();
  RpcDeniedReply reply = new RpcDeniedReply(call.getXid(),
      RpcReply.ReplyState.MSG_DENIED,
      RpcDeniedReply.RejectState.AUTH_ERROR, new VerifierNone());
  reply.write(out);
  ChannelBuffer buf = ChannelBuffers.wrappedBuffer(out.asReadOnlyWrap()
      .buffer());
  RpcResponse rsp = new RpcResponse(buf, remoteAddress);
  RpcUtil.sendRpcResponse(ctx, rsp);
}
 

public static String socketAddress2String(final SocketAddress addr) {
    StringBuilder sb = new StringBuilder();
    InetSocketAddress inetSocketAddress = (InetSocketAddress)addr;
    sb.append(inetSocketAddress.getAddress().getHostAddress());
    sb.append(":");
    sb.append(inetSocketAddress.getPort());
    return sb.toString();
}