import org.jboss.netty.handler.codec.http.HttpResponseDecoder; //导入依赖的package包/类
@Override
public ChannelPipeline getPipeline() throws Exception {
    ChannelPipeline pipeline = pipeline();

    pipeline.addLast("idle-detection", IDLE_STATE_HANDLER);
    pipeline.addLast("logging", new LoggingHandler(InternalLogLevel.DEBUG));
    pipeline.addLast("http-deflater", new HttpContentCompressor());
    pipeline.addLast("decoder", new HttpResponseDecoder(RESPONSE_MAX_INITIAL_LINE_LENGTH, RESPONSE_MAX_HEADER_SIZE, RESPONSE_MAX_CHUNK_SIZE));
    pipeline.addLast("encoder", new HttpRequestEncoder());
    pipeline.addLast("http-inflater", new HttpContentDecompressor());
    pipeline.addLast("remote-hop-timer", new ClientTimingHandler("outbound"));
    pipeline.addLast("exception-surfacer", EXCEPTION_SURFACER);
    pipeline.addLast("idle-watchdog", new IdleChannelWatchdog("outbound"));

    return pipeline;
}
 

import org.jboss.netty.handler.codec.http.HttpResponseDecoder; //导入依赖的package包/类
private void initSendPipeline(ChannelPipeline pipeline) {
	pipeline.addLast("reqencoder", new HttpRequestEncoder()); // downstream
	pipeline.addLast("respdecoder", new HttpResponseDecoder()); // upstream
	pipeline.addLast("aggregator", new HttpChunkAggregator(HttpTunnelMessageUtils.MAX_BODY_SIZE)); // upstream
	pipeline.addLast(HttpTunnelClientChannelProxyHandler.NAME, sendHttpHandler); // proxy auth, etc
	pipeline.addLast(HttpTunnelClientChannelSendHandler.NAME, sendHandler); // both
	pipeline.addLast("writeFragmenter", new WriteFragmenter(HttpTunnelMessageUtils.MAX_BODY_SIZE));
}
 

import org.jboss.netty.handler.codec.http.HttpResponseDecoder; //导入依赖的package包/类
private void initPollPipeline(ChannelPipeline pipeline) {
	pipeline.addLast("reqencoder", new HttpRequestEncoder()); // downstream
	pipeline.addLast("respdecoder", new HttpResponseDecoder()); // upstream
	pipeline.addLast("aggregator", new HttpChunkAggregator(HttpTunnelMessageUtils.MAX_BODY_SIZE)); // upstream
	pipeline.addLast(HttpTunnelClientChannelProxyHandler.NAME, pollHttpHandler); // proxy auth, etc
	pipeline.addLast(HttpTunnelClientChannelPollHandler.NAME, pollHandler); // both
}
 

import org.jboss.netty.handler.codec.http.HttpResponseDecoder; //导入依赖的package包/类
public static ClientChannel connect(LoggerFactory loggerFactory, URI uri,
        int connectTimeoutMillis) throws ChannelException {
    Logger logger = loggerFactory.create(String.format("WebSocketClientHandler-%s", uri));

    WebSocketClientChannel clientChannel = new WebSocketClientChannel();
    clientChannel.setUri(uri);

    ConnectingChannelHandler handler = new ConnectingChannelHandler();
    clientChannel.setChannelHandler(handler);

    WebSocketClientUpstreamHandler wsHandler = new WebSocketClientUpstreamHandler(logger,
            clientChannel);
    ChannelPipeline pipeline = Channels.pipeline();
    pipeline.addLast("decoder", new HttpResponseDecoder());
    pipeline.addLast("encoder", new HttpRequestEncoder());
    // connect
    Channel channel = prepareAndConnect(logger, uri, pipeline, wsHandler,
            uri.getScheme().equalsIgnoreCase("wss"), connectTimeoutMillis);
    // handshake
    try {
        WebSocketClientHandshaker handshaker = wsFactory.newHandshaker(uri,
                WebSocketVersion.V13, null, true, WebSocketClientHelper.getHeaders(uri));
        wsHandler.handshaker = handshaker;
        handshaker.handshake(channel);
        // return maybe fast than call
        if (!wsHandler.handshaker.isHandshakeComplete() && (handler.error == null)) {
            synchronized (handler.syncObject) {
                handler.syncObject.wait(connectTimeoutMillis);
            }
        }
    } catch (Exception e) {
        throw new ChannelException(Text.WS_HANDSHAKE_ERROR, e);
    }

    if (wsHandler.handshaker.isHandshakeComplete()) {
        return clientChannel;
    }
    if (handler.error != null) {
        throw new ChannelException(Text.CONNECT_FAIL + ": " + handler.error.getMessage(),
                handler.error);
    }

    throw new ChannelException(Text.CONNECT_TIMEOUT);
}
 

import org.jboss.netty.handler.codec.http.HttpResponseDecoder; //导入依赖的package包/类
public NettyWSClient(String uriStr, WebSocketClient webSocketClient)
{
	
	this.t = webSocketClient;
	this.uri = URI.create(uriStr);
	

	String protocol = this.uri.getScheme();
	if (!protocol.equals("ws"))
	{
		throw new IllegalArgumentException("Unsupported protocol: " + protocol);
	}
	
	ClientBootstrap bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool()));
	
	Channel ch = null;

	HashMap<String, String> customHeaders = new HashMap<String, String>();
	customHeaders.put("MyHeader", "MyValue");
	
	// Connect with V13 (RFC 6455 aka HyBi-17). You can change it to V08 or V00.
	// If you change it to V00, ping is not supported and remember to change
	// HttpResponseDecoder to WebSocketHttpResponseDecoder in the pipeline.
	this.handshaker = new WebSocketClientHandshakerFactory().newHandshaker(this.uri, WebSocketVersion.V13, null, false, customHeaders);
	
	bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
		public ChannelPipeline getPipeline() throws Exception
		{
			ChannelPipeline pipeline = Channels.pipeline();
			
			pipeline.addLast("decoder", new HttpResponseDecoder());
			pipeline.addLast("encoder", new HttpRequestEncoder());
			pipeline.addLast("ws-handler", NettyWSClient.this);
			return pipeline;
		}
	});
	
	ChannelFuture future = bootstrap.connect(new InetSocketAddress(this.uri.getHost(), this.uri.getPort()));
	future.syncUninterruptibly();
	ch = future.getChannel();
	try
	{
		this.handshaker.handshake(ch);
	}
	catch (Exception e)
	{
		// TODO Auto-generated catch block
		e.printStackTrace();
	}
}