目錄
- [HttpRequsetHandler][1] - 管理HTTP請求和響應
- TextWebSocketFrameHandler[2] - 處理聊天的WebSocket幀
- ChatForWebSocketServerInitailizer [3] - 初始化PipelineChannel
- ChatWebSocketServer[4]-啟動程序
- 增加SSL- SecurityChatWebSocketServerInitailizer [5]
- 增加SSL- SecurityChatWebSocketServer[6]
HttpRequsetHandler
import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.stream.ChunkedNioFile;
import java.io.File;
import java.io.RandomAccessFile;
import java.net.URISyntaxException;
import java.net.URL;
public class HttpRequsetHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
private final String wsUri;
private static File INDEX;
static{
URL location = HttpRequsetHandler.class.getProtectionDomain().getCodeSource().getLocation();
try{
String path = location.toURI() + "index.html";
path = !path.contains("file:")?path:path.substring(5);
INDEX = new File(path);
}catch (URISyntaxException e){
e.printStackTrace();;
}
}
public HttpRequsetHandler(String wsUri) {
this.wsUri = wsUri;
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest request) throws Exception {
if (wsUri.equalsIgnoreCase(request.getUri())){
//如果請求了websocket協(xié)議升級斗蒋，則增加引用計數(shù)(調用retain()方法),并傳給下一個ChannelInboundHandler
ctx.fireChannelRead(request.retain());
}else{
if (HttpHeaders.is100ContinueExpected(request)){ //讀取is100ContinueExpected請求，符合http1.1版本
send100Continue(ctx);
}
RandomAccessFile file = new RandomAccessFile(INDEX,"r");
HttpResponse response = new DefaultFullHttpResponse(
request.getProtocolVersion(), HttpResponseStatus.OK
);
response.headers().set(
HttpHeaders.Names.CONTENT_TYPE,"text/html;charset=UTF-8"
);
boolean keepAlive = HttpHeaders.isKeepAlive(request);
if (keepAlive){ //如果請求了keep-alive妇押，則添加所需要的HTTP頭信息
response.headers().set(
HttpHeaders.Names.CONTENT_LENGTH,file.length()
);
response.headers().set(
HttpHeaders.Names.CONNECTION,HttpHeaders.Values.KEEP_ALIVE
);
}
ctx.write(response);//將httpresponse寫入到客戶端
if (ctx.pipeline().get(SslHandler.class)==null){//將index.html寫到客戶端
ctx.write(new DefaultFileRegion(file.getChannel(),0,file.length()));
}else{
ctx.write(new ChunkedNioFile(file.getChannel()));
}
ChannelFuture future = ctx.writeAndFlush(LastHttpContent.EMPTY_LAST_CONTENT);
if (!keepAlive){
future.addListener(
ChannelFutureListener.CLOSE
);
}
}
}
private void send100Continue(ChannelHandlerContext ctx) {
FullHttpResponse response = new DefaultFullHttpResponse(
HttpVersion.HTTP_1_0,HttpResponseStatus.CONTINUE
);
ctx.writeAndFlush(response);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
cause.printStackTrace();
ctx.close();
}
}
TextWebSocketFrameHandler-處理聊天的WebSocket幀
public class TextWebSocketFrameHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
private final ChannelGroup group;
public TextWebSocketFrameHandler(ChannelGroup group) {
this.group = group;
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt== WebSocketServerProtocolHandler.ServerHandshakeStateEvent.HANDSHAKE_COMPLETE){
//握手成功俊马，則從該ChannelPipeline移出HttpRequsetHandler，因為它將接收不到任何消息了
ctx.pipeline().remove(HttpRequsetHandler.class);//
//通知所有已連接的WebSocket客戶端解寝，新客戶端已上線
group.writeAndFlush(new TextWebSocketFrame("CLient "+ctx.channel()) +"joined");
group.add(ctx.channel());
}else{
super.userEventTriggered(ctx,evt);
}
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
group.writeAndFlush(msg.retain());//增加消息的引用計數(shù)聋伦，并將它寫到ChannelGroup中所有已連接的客戶端
}
}
初始化PipelineChannel
public class ChatForWebSocketServerInitailizer extends ChannelInitializer<Channel> {
private final ChannelGroup group;
public ChatForWebSocketServerInitailizer(ChannelGroup congrouptext) {
this.group = congrouptext;
}
@Override
protected void initChannel(Channel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(
new HttpServerCodec(), //字節(jié)解碼為httpreqeust,httpcontent,last-httpcontent
new ChunkedWriteHandler(),//寫入一個文本
new HttpObjectAggregator(64*1024), //HttpMessage與多個HttpConent聚合
new HttpRequsetHandler("/ws"), //處理FullJttpRequest——不往/ws上發(fā)的請求
new WebSocketServerProtocolHandler("/ws"), //處理websocket升級握手嘉抓、PingWebSocketFrame抑片、PongWebSocketFrame、CloseWebSocketFrame
new TextWebSocketFrameHandler(group), //處理TextWebSocketFrameHandler和握手完成事件,如握手成功敞斋，則所需要的Handler將會被添加到ChannelPipeline植捎，不需要的則會移除
);//如果請求端是websocket則升級該握手
}
}
ChatWebSocketServer-啟動程序
public class ChatWebSocketServer {
//創(chuàng)建DefaultChannelGroup焰枢，將其保存為所有已經連接的WebSocket Channel
private final ChannelGroup channelGroup = new DefaultChannelGroup(ImmediateEventExecutor.INSTANCE);
private final EventLoopGroup group = new NioEventLoopGroup();
private Channel channel;
public ChannelFuture start(InetSocketAddress address){
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(group)
.channel(NioServerSocketChannel.class)
.childHandler(createInitializer(channelGroup));
ChannelFuture future = bootstrap.bind(address);
future.syncUninterruptibly();
channel = future.channel();
return future;
}
protected ChannelHandler createInitializer(ChannelGroup channelGroup) {
return new ChatWebSocketServerInitailizer(channelGroup);
}
public void destory(){
if (channel!=null){
channel.close();
}
channelGroup.close();
group.shutdownGracefully();
}
public static void main(String[] args) {
if (args.length!=1){
System.out.print("Please give port as argument!");
System.exit(1);
}
int port = Integer.parseInt(args[0]);
final ChatWebSocketServer endpoint = new ChatWebSocketServer();
ChannelFuture future = endpoint.start(new InetSocketAddress(port));
Runtime.getRuntime().addShutdownHook(new Thread(){
@Override
public void run(){
endpoint.destory();
}
});
future.channel().closeFuture().syncUninterruptibly();
}
}
加密
SecurityChatWebSocketServerInitailizer
public class SecurityChatWebSocketServerInitailizer extends ChatWebSocketServerInitailizer {
private final SslContext context;
public SecurityChatWebSocketServerInitailizer(ChannelGroup congrouptext, SslContext context) {
super(congrouptext);
this.context = context;
}
@Override
protected void initChannel(Channel ch) throws Exception{
super.initChannel(ch);
SSLEngine engine = context.newEngine(ch.alloc());
engine.setUseClientMode(false);
ch.pipeline().addFirst(new SslHandler(engine)); //將SslHandler添加到ChannelPipeline中
}
}
啟動程序 - SecurityChatWebSocketServer
public class SecurityChatWebSocketServer extends ChatWebSocketServer{
private final SslContext context;
public SecurityChatWebSocketServer(SslContext context) {
this.context = context;
}
public ChannelFuture start(InetSocketAddress address){
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(group)
.channel(NioServerSocketChannel.class)
.childHandler(createInitializer(channelGroup));
ChannelFuture future = bootstrap.bind(address);
future.syncUninterruptibly();
channel = future.channel();
return future;
}
protected ChannelHandler createInitializer(ChannelGroup channelGroup) {
//return new ChatWebSocketServerInitailizer(channelGroup);
return new SecurityChatWebSocketServerInitailizer(channelGroup,context);
}
public void destory(){
if (channel!=null){
channel.close();
}
channelGroup.close();
group.shutdownGracefully();
}
public static void main(String[] args) throws SSLException, CertificateException {
if (args.length!=1){
System.out.print("Please give port as argument!");
System.exit(1);
}
int port = Integer.parseInt(args[0]);
SelfSignedCertificate cert = new SelfSignedCertificate();
SslContext context = SslContext.newServerContext(cert.certificate(),cert.privateKey());
final SecurityChatWebSocketServer endpoint = new SecurityChatWebSocketServer(context);
ChannelFuture future = endpoint.start(new InetSocketAddress(port));
Runtime.getRuntime().addShutdownHook(new Thread(){
@Override
public void run(){
endpoint.destory();
}
});
future.channel().closeFuture().syncUninterruptibly();
}
}
