Java 类org.jboss.netty.channel.socket.SocketChannel 实例源码

@Override
public SocketChannel newChannel(ChannelPipeline pipeline) {
  try {
    SSLContext sslContext = SSLContext.getInstance("TLS");
    sslContext.init(null, new TrustManager[]{new PermissiveTrustManager()},
                    null);
    SSLEngine sslEngine = sslContext.createSSLEngine();
    sslEngine.setUseClientMode(true);
    // addFirst() will make SSL handling the first stage of decoding
    // and the last stage of encoding
    pipeline.addFirst("ssl", new SslHandler(sslEngine));
    return super.newChannel(pipeline);
  } catch (Exception ex) {
    throw new RuntimeException("Cannot create SSL channel", ex);
  }
}

/**
 * everytime the timer fires run this!
 * Looks through every proxy and determines if the proxy needs to 
 * attempt to connect to the controller.
 */
public synchronized void run(){
    log.debug("Looking for controllers not currently connected");
    Iterator <Long> it = proxies.keySet().iterator();
    while(it.hasNext()){
        List <Proxy> ps = proxies.get(it.next());
        Iterator <Proxy> proxyIt = ps.iterator();
        while(proxyIt.hasNext()){
            Proxy p = proxyIt.next();
            log.debug("Proxy for " + p.getSwitch().getStringId() + " " + p.getSlicer().getControllerAddress().toString() + " is connected: " + p.connected());
            if(!p.connected() && p.getAdminStatus()){
                log.debug("Creating new Channel to " + p.getSlicer().getControllerAddress().toString() + " for switch: " + p.getSwitch().getStringId());
                SocketChannel controller_channel = channelCreator.newChannel(getPipeline());
                p.connect(controller_channel);
            }
        }

    }
}

@Override
public SocketChannel newChannel(ChannelPipeline pipeline) {
  try {

    ZlibEncoder encoder = new ZlibEncoder(compressionLevel);
    pipeline.addFirst("deflater", encoder);
    pipeline.addFirst("inflater", new ZlibDecoder());
    return super.newChannel(pipeline);
  } catch (Exception ex) {
    throw new RuntimeException("Cannot create Compression channel", ex);
  }
}

@Override
protected Shuffle getShuffle(final Configuration conf) {
  return new Shuffle(conf) {
    @Override
    protected void verifyRequest(String appid, ChannelHandlerContext ctx,
        HttpRequest request, HttpResponse response, URL requestUri)
        throws IOException {
      SocketChannel channel = (SocketChannel)(ctx.getChannel());
      socketKeepAlive = channel.getConfig().isKeepAlive();
    }
  };
}

@Override
protected Shuffle getShuffle(final Configuration conf) {
  return new Shuffle(conf) {
    @Override
    protected void verifyRequest(String appid, ChannelHandlerContext ctx,
        HttpRequest request, HttpResponse response, URL requestUri)
        throws IOException {
      SocketChannel channel = (SocketChannel)(ctx.getChannel());
      socketKeepAlive = channel.getConfig().isKeepAlive();
    }
  };
}

public void setupChannel() throws IOException{
    ChannelFuture future = createMock(org.jboss.netty.channel.ChannelFuture.class);
    ChannelPipeline pipeline = createMock(org.jboss.netty.channel.ChannelPipeline.class);
    ChannelHandlerContext context = createMock(org.jboss.netty.channel.ChannelHandlerContext.class);
    handler = EasyMock.createNiceMock(edu.iu.grnoc.flowspace_firewall.OFControllerChannelHandler.class);
    channel = EasyMock.createNiceMock(org.jboss.netty.channel.socket.SocketChannel.class);

    ChannelFuture otherFuture = createMock(org.jboss.netty.channel.ChannelFuture.class);
    expect(channel.getPipeline()).andReturn(pipeline).anyTimes();
    expect(pipeline.getContext("handler")).andReturn(context).anyTimes();
    expect(context.getHandler()).andReturn(handler).anyTimes();
    expect(channel.connect(EasyMock.isA(java.net.InetSocketAddress.class))).andReturn(future).anyTimes();
    expect(channel.write(EasyMock.isA(org.openflow.protocol.OFMessage.class))).andReturn(otherFuture).anyTimes();

    handler.setSwitch(EasyMock.isA(net.floodlightcontroller.core.IOFSwitch.class));
    EasyMock.expectLastCall().anyTimes();

    handler.setProxy(EasyMock.isA(edu.iu.grnoc.flowspace_firewall.Proxy.class));
    EasyMock.expectLastCall().anyTimes();

    handler.sendMessage(EasyMock.isA(org.openflow.protocol.OFMessage.class));
    EasyMock.expectLastCall().andAnswer(new IAnswer<Object>() {
        public Object answer() {
            //supply your mock implementation here...
            messagesSentToController.add((OFMessage)EasyMock.getCurrentArguments()[0]);
            //return the value to be returned by the method (null for void)
            return null;
        }
    }).anyTimes();


    EasyMock.replay(future);
    EasyMock.replay(pipeline);
    EasyMock.replay(context);
    //EasyMock.replay(handler);
    EasyMock.replay(otherFuture);
}

@Override
public SocketChannel newChannel(ChannelPipeline pipeline) {
    try {
        SSLContext sslContext = SSLContext.getInstance("TLSv1.2");
        sslContext.init(null, new TrustManager[]{new BogusTrustManager(publicKey)},
                null);
        SSLEngine sslEngine = sslContext.createSSLEngine();
        sslEngine.setUseClientMode(true);
        pipeline.addFirst("ssl", new SslHandler(sslEngine));
        return super.newChannel(pipeline);
    } catch (Exception ex) {
        throw new RuntimeException("Cannot create SSL channel", ex);
    }
}

@Override
protected Shuffle getShuffle(final Configuration conf) {
  return new Shuffle(conf) {
    @Override
    protected void verifyRequest(String appid, ChannelHandlerContext ctx,
        HttpRequest request, HttpResponse response, URL requestUri)
        throws IOException {
      SocketChannel channel = (SocketChannel)(ctx.getChannel());
      socketKeepAlive = channel.getConfig().isKeepAlive();
    }
  };
}

private SocketChannel createClientChannel() {
    InetSocketAddress serverAddress = new InetSocketAddress("localhost",
            SERVER_PORT);
    ChannelFuture clientChannelFuture = clientBootstrap
            .connect(serverAddress);
    try {
        if (!clientChannelFuture.await(1000, TimeUnit.MILLISECONDS)) {
            LOG.severe("did not connect within acceptable time period");
            return null;
        }
    } catch (InterruptedException e) {
        LOG.severe("Interrupted while waiting for client connect to be established");
        return null;
    }

    if (!clientChannelFuture.isSuccess()) {
        LOG.log(Level.SEVERE, "did not connect successfully",
                clientChannelFuture.getCause());
        return null;
    }

    HttpTunnelClientChannelConfig config = (HttpTunnelClientChannelConfig) clientChannelFuture
            .getChannel().getConfig();
    config.setWriteBufferHighWaterMark(2 * 1024 * 1024);
    config.setWriteBufferLowWaterMark(1024 * 1024);

    return (SocketChannel) clientChannelFuture.getChannel();
}

@Override
public SocketChannel newChannel(ChannelPipeline pipeline) {
    FakeSocketChannel channel = new FakeSocketChannel(null, this, pipeline,
            new FakeChannelSink());
    createdChannels.add(channel);
    return channel;
}

@Override
public SocketChannel newChannel(ChannelPipeline pipeline) {
  TrustManager[] managers;
  try {
    if (enableCompression) {
      ZlibEncoder encoder = new ZlibEncoder(compressionLevel);
      pipeline.addFirst("deflater", encoder);
      pipeline.addFirst("inflater", new ZlibDecoder());
    }
    if (enableSsl) {
      if (trustAllCerts) {
        logger.warn("No truststore configured, setting TrustManager to accept"
            + " all server certificates");
        managers = new TrustManager[] { new PermissiveTrustManager() };
      } else {
        KeyStore keystore = null;

        if (truststore != null) {
          if (truststorePassword == null) {
            throw new NullPointerException("truststore password is null");
          }
          InputStream truststoreStream = new FileInputStream(truststore);
          keystore = KeyStore.getInstance(truststoreType);
          keystore.load(truststoreStream, truststorePassword.toCharArray());
        }

        TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509");
        // null keystore is OK, with SunX509 it defaults to system CA Certs
        // see http://docs.oracle.com/javase/6/docs/technotes/guides/security/jsse/JSSERefGuide.html#X509TrustManager
        tmf.init(keystore);
        managers = tmf.getTrustManagers();
      }

      SSLContext sslContext = SSLContext.getInstance("TLS");
      sslContext.init(null, managers, null);
      SSLEngine sslEngine = sslContext.createSSLEngine();
      sslEngine.setUseClientMode(true);
      List<String> enabledProtocols = new ArrayList<String>();
      for (String protocol : sslEngine.getEnabledProtocols()) {
        if (!excludeProtocols.contains(protocol)) {
          enabledProtocols.add(protocol);
        }
      }
      sslEngine.setEnabledProtocols(enabledProtocols.toArray(new String[0]));
      logger.info("SSLEngine protocols enabled: " +
          Arrays.asList(sslEngine.getEnabledProtocols()));
      // addFirst() will make SSL handling the first stage of decoding
      // and the last stage of encoding this must be added after
      // adding compression handling above
      pipeline.addFirst("ssl", new SslHandler(sslEngine));
    }

    return super.newChannel(pipeline);
  } catch (Exception ex) {
    logger.error("Cannot create SSL channel", ex);
    throw new RuntimeException("Cannot create SSL channel", ex);
  }
}

public SocketChannel newChannel(ChannelPipeline pipeline) {
    return new NioClientSocketChannel(this, pipeline, sink, workerPool.nextWorker());
}

public SocketChannel newChannel(ChannelPipeline pipeline) {
    return new NioClientSocketChannel(this, pipeline, sink, workerPool.nextWorker());
}

public void run() throws InterruptedException {
    LOG.info("binding server channel");
    Channel serverChannel = serverBootstrap.bind(new InetSocketAddress(
            SERVER_PORT));
    channels.add(serverChannel);
    LOG.log(Level.INFO, "server channel bound to {0}",
            serverChannel.getLocalAddress());

    SocketChannel clientChannel = createClientChannel();
    if (clientChannel == null) {
        LOG.severe("no client channel - bailing out");
        return;
    }

    channels.add(clientChannel);
    c2sDataSender.setChannel(clientChannel);

    executor.execute(c2sDataSender);

    if (!c2sDataSender.waitForFinish(5, TimeUnit.MINUTES)) {
        LOG.severe("Data send from client to server failed");
    }

    if (!s2cDataSender.waitForFinish(5, TimeUnit.MINUTES)) {
        LOG.severe("Data send from server to client failed");
    }

    LOG.log(Level.INFO, "Waiting for verification to complete");
    if (!c2sVerifier.waitForCompletion(30L, TimeUnit.SECONDS)) {
        LOG.warning("Timed out waiting for verification of client-to-server stream");
    }

    if (!s2cVerifier.waitForCompletion(30L, TimeUnit.SECONDS)) {
        LOG.warning("Timed out waiting for verification of server-to-client stream");
    }

    LOG.info("closing client channel");
    closeChannel(clientChannel);
    LOG.info("server channel status: "
            + (serverChannel.isOpen() ? "open" : "closed"));
    LOG.info("closing server channel");
    closeChannel(serverChannel);
}