目录
- [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协议升级，则增加引用计数(调用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());//增加消息的引用计数，并将它写到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(), //字节解码为httpreqeust,httpcontent,last-httpcontent
new ChunkedWriteHandler(),//写入一个文本
new HttpObjectAggregator(64*1024), //HttpMessage与多个HttpConent聚合
new HttpRequsetHandler("/ws"), //处理FullJttpRequest——不往/ws上发的请求
new WebSocketServerProtocolHandler("/ws"), //处理websocket升级握手、PingWebSocketFrame、PongWebSocketFrame、CloseWebSocketFrame
new TextWebSocketFrameHandler(group), //处理TextWebSocketFrameHandler和握手完成事件,如握手成功，则所需要的Handler将会被添加到ChannelPipeline，不需要的则会移除
);//如果请求端是websocket则升级该握手
}
}
ChatWebSocketServer-启动程序
public class ChatWebSocketServer {
//创建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();
}
}
网友评论