facebook/swift:构建thrift http server(4)--ThriftXHRDecoder,ThriftXHREncoder

Dear 丶 2022-02-05 04:49 160阅读 0赞

[《facebook/swift:构建thrift http server(1)》][facebook_swift_thrift http server_1]  
[《facebook/swift:构建thrift http server(2)–HttpServerCodec》][facebook_swift_thrift http server_2_HttpServerCodec]  
[《facebook/swift:构建thrift http server(3)–CORS跨域》][facebook_swift_thrift http server_3_CORS]

在上一篇博客中解决了thrift http sever的CORS跨域问题，但前端依然没有服务端的正常响应。看来还存在问题。  
继续研究Netty的代码。

## ThriftMessage ##

通过跟踪服务端收到的HTTP POST请求在管道(ChannelPipeline)中的传递流程找到了问题：  
如下是实现将网络请求分发到thrift服务实例(NiftyProcessor)的ChannelHandler实例[com.facebook.nifty.core.NiftyDispatcher][]的`messageReceived`方法实现代码：

public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
                throws Exception
        {
            	
            if (e.getMessage() instanceof ThriftMessage) {
        
                ThriftMessage message = (ThriftMessage) e.getMessage();
                if (taskTimeoutMillis > 0) {
        
                    message.setProcessStartTimeMillis(System.currentTimeMillis());
                }
                checkResponseOrderingRequirements(ctx, message);
    
                TNiftyTransport messageTransport = new TNiftyTransport(ctx.getChannel(), message);
                TTransportPair transportPair = TTransportPair.fromSingleTransport(messageTransport);
                TProtocolPair protocolPair = duplexProtocolFactory.getProtocolPair(transportPair);
                TProtocol inProtocol = protocolPair.getInputProtocol();
                TProtocol outProtocol = protocolPair.getOutputProtocol();
    
                processRequest(ctx, message, messageTransport, inProtocol, outProtocol);
            }
            else {
        
            	// 如果不是message是ThriftMessage实例则继续向下传递
                ctx.sendUpstream(e);
            }
        }

请注意第一行的判断语句，只有收到的消息是[ThriftMessage][]实例，才会响应请求。  
而[HttpServerCodec][]只会将收到的网络请求解析为[DefaultHttpRequest][]。

> 参见 [org.jboss.netty.handler.codec.http.HttpRequestDecoder][]

所以问题搞清楚了。我们需要一个`ChannelUpstreamHandler`将`DefaultHttpRequest`转换为`ThriftMessage`向后传递给`NiftyDispatcher`,同时也需要一个`ChannelDownstreamHandler`将`ThriftMessage`转为`DefaultHttpResponse`向前传递给`HttpServerCodec`由`HttpResponseEncoder`进一步封装成HTTP响应数据最后发送给前端。

## ThriftXHRDecoder ##

HTTP request解码器ThriftXHRDecoder实现如下:

import org.jboss.netty.channel.ChannelHandlerContext;
    import org.jboss.netty.channel.MessageEvent;
    import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
    import org.jboss.netty.channel.UpstreamMessageEvent;
    import org.jboss.netty.handler.codec.http.HttpMethod;
    import org.jboss.netty.handler.codec.http.HttpRequest;
    
    import com.facebook.nifty.core.ThriftMessage;
    import com.facebook.nifty.core.ThriftTransportType;
    
    /**
     * XHR(XML Http Request)解码器<br>
     * 将{@link HttpRequest}请求的内容数据(content)转为{@link ThriftMessage},
     * 提供给{@link com.facebook.nifty.core.NiftyDispatcher}
     * @author guyadong
     *
     */
    public class ThriftXHRDecoder extends SimpleChannelUpstreamHandler {
        
    	
    	public ThriftXHRDecoder() {
        
    	}
    
    	@Override
    	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        
    		if(e.getMessage() instanceof HttpRequest){
        
    			HttpRequest request = (HttpRequest)e.getMessage();
    			if(request.getContent().readable() && HttpMethod.POST.equals(request.getMethod())){
        
    				// 非FRAME数据
    				ThriftMessage thriftMessage = new ThriftMessage(request.getContent(),ThriftTransportType.UNFRAMED);
    				ctx.sendUpstream(new UpstreamMessageEvent(ctx.getChannel(), thriftMessage, e.getRemoteAddress()));
    				return;
    			}
    		}
    		super.messageReceived(ctx, e);
    	}
    }

## ThriftXHREncoder ##

`ThriftMessage`编码器ThriftXHREncoder实现如下：

import org.jboss.netty.channel.ChannelHandlerContext;
    import org.jboss.netty.channel.Channels;
    import org.jboss.netty.channel.DownstreamMessageEvent;
    import org.jboss.netty.channel.MessageEvent;
    import org.jboss.netty.channel.SimpleChannelDownstreamHandler;
    import org.jboss.netty.handler.codec.http.DefaultHttpResponse;
    import org.jboss.netty.handler.codec.http.HttpResponseStatus;
    
    import com.facebook.nifty.core.ThriftMessage;
    import static org.jboss.netty.handler.codec.http.HttpVersion.HTTP_1_1;
    
    /**
     * XHR(XML Http Request)编码器<br>
     * 将{@link com.facebook.nifty.core.NiftyDispatcher}输出的
     * {@link ThriftMessage}响应数据转为{@link DownstreamMessageEvent},
     * 
     * @author guyadong
     *
     */
    public class ThriftXHREncoder extends SimpleChannelDownstreamHandler {
        
    
    	@Override
    	public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
        
    		if(e.getMessage() instanceof ThriftMessage){
        
    			ThriftMessage thriftMessage = (ThriftMessage)e.getMessage();
    			if(thriftMessage.getBuffer().readable()){
        
    				switch (thriftMessage.getTransportType()) {
        
    				case UNFRAMED:
                        DefaultHttpResponse response = new DefaultHttpResponse(HTTP_1_1, HttpResponseStatus.OK);
                        response.setContent(thriftMessage.getBuffer());
    					ctx.sendDownstream(new DownstreamMessageEvent(ctx.getChannel(), 
    							Channels.future(ctx.getChannel()), response, e.getRemoteAddress()));
    					return;
    				default:
    	                throw new UnsupportedOperationException(
    	                		thriftMessage.getTransportType().name() +" transport is not supported");
    				}
    			}
    		}
    		super.writeRequested(ctx, e);
    	}
    
    }

## 修改ChannelPipeline ##

有了`ThriftXHRDecoder`和`ThriftXHREncoder`就可以将它们添加到`frameCodec`之后，`dispatcher`之前，基于上一篇中`addCorsHandlerIfHttp`方法代码修改如下，

/**
    	 * 添加CORS Handler和XHR编解码器
    	 */
    	protected void addCorsHandlerIfHttp(){
        
    		if(HTTP_TRANSPORT.equals(thriftServerConfig.getTransportName())){
        
    			try {
        
    				// 反射获取私有的成员NettyServerTransport
    				final NettyServerTransport nettyServerTransport = ReflectionUtils.valueOfField(thriftServer, "transport");
    				// 反射获取私有的成员ChannelPipelineFactory
    				Field pipelineFactory = NettyServerTransport.class.getDeclaredField("pipelineFactory");
    				{
        
    					Field modifiersField = Field.class.getDeclaredField("modifiers");
    					modifiersField.setAccessible(true); //Field 的 modifiers 是私有的
    					modifiersField.setInt(pipelineFactory, pipelineFactory.getModifiers() & ~Modifier.FINAL);
    				}
    				pipelineFactory.setAccessible(true);
    				final ChannelPipelineFactory channelPipelineFactory = (ChannelPipelineFactory) pipelineFactory.get(nettyServerTransport);
    				final Netty3CorsConfig corsConfig = Netty3CorsConfigBuilder.forAnyOrigin()
    					.allowedRequestMethods(POST,GET,OPTIONS)
    					.allowedRequestHeaders("Origin","Content-Type","Accept","application","x-requested-with")
    					.build();
    				ChannelPipelineFactory factoryWithCORS = new ChannelPipelineFactory(){
        
    
    					@Override
    					public ChannelPipeline getPipeline() throws Exception {
        
    						// 修改 ChannelPipeline,在frameCodec后(顺序)增加CORS handler,XHR编解码器
    						ChannelPipeline cp = channelPipelineFactory.getPipeline();
    						cp.addAfter("frameCodec", "thriftXHRDecoder", new ThriftXHRDecoder());
    						cp.addAfter("frameCodec", "thriftXHREncoder", new ThriftXHREncoder());
    						cp.addAfter("frameCodec", "cors", new Netty3CorsHandler(corsConfig));
    						return cp;
    					}};
    				// 修改nettyServerTransport的私有常量pipelineFactory
    				pipelineFactory.set(nettyServerTransport, factoryWithCORS);
    			} catch (Exception e) {
        
    				Throwables.throwIfUnchecked(e);
    				throw new RuntimeException(e);
    			}
    		}
    	}

修改完毕后再执行test\_js.html,thrift http server终于有响应了。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0LzEwa20_size_16_color_FFFFFF_t_70]  
而且数据结果正常  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0LzEwa20_size_16_color_FFFFFF_t_70 1]

## 遗留问题 ##

至此，基于facebook/swift构建的thrift http server已经基本可以正常工作，但还是存在一个小问题。HTTP响应只会在前端空闲超时后才会发送响应数据到前端，所以`ThriftServerConfig`的IdleConnectionTimeout如果设置过大(比如默认值60s)，那么前端要在发送请求60秒后才会收到服务端的响应，这个问题一直到目前还没有找到。所以目前的办法是将这个值设置在10ms，就基本不会影响前端的响应延迟。

零零散散的写了好几篇文章，贴出的代码都比较零碎，完整的代码参见码云仓库  
[https://gitee.com/l0km/common-java/tree/master/common-thrift/src/main/java/com/facebook/swift/service][https_gitee.com_l0km_common-java_tree_master_common-thrift_src_main_java_com_facebook_swift_service]

重要的修改在[ThriftServerService][]

[facebook_swift_thrift http server_1]: https://blog.csdn.net/10km/article/details/89792184
[facebook_swift_thrift http server_2_HttpServerCodec]: https://blog.csdn.net/10km/article/details/89810011
[facebook_swift_thrift http server_3_CORS]: https://blog.csdn.net/10km/article/details/89816763
[com.facebook.nifty.core.NiftyDispatcher]: https://github.com/facebookarchive/nifty/blob/nifty-parent-0.14.1/nifty-core/src/main/java/com/facebook/nifty/core/NiftyDispatcher.java
[ThriftMessage]: https://github.com/facebookarchive/nifty/blob/nifty-parent-0.14.1/nifty-core/src/main/java/com/facebook/nifty/core/ThriftMessage.java
[HttpServerCodec]: https://github.com/netty/netty/blob/3.7/src/main/java/org/jboss/netty/handler/codec/http/HttpServerCodec.java
[DefaultHttpRequest]: https://github.com/netty/netty/blob/3.7/src/main/java/org/jboss/netty/handler/codec/http/DefaultHttpRequest.java
[org.jboss.netty.handler.codec.http.HttpRequestDecoder]: https://github.com/netty/netty/blob/3.7/src/main/java/org/jboss/netty/handler/codec/http/HttpRequestDecoder.java
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0LzEwa20_size_16_color_FFFFFF_t_70]: /images/20220205/33ee80db030544f6811e176b4e4be22c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0LzEwa20_size_16_color_FFFFFF_t_70 1]: /images/20220205/3c2ce68def784b53afb148d28909bb85.png
[https_gitee.com_l0km_common-java_tree_master_common-thrift_src_main_java_com_facebook_swift_service]: https://gitee.com/l0km/common-java/tree/master/common-thrift/src/main/java/com/facebook/swift/service
[ThriftServerService]: https://gitee.com/l0km/common-java/blob/master/common-thrift/src/main/java/com/facebook/swift/service/ThriftServerService.java