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

    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);
        }
    }

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);
	}
}

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);
	}

}

	/**
	 * 添加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);
			}
		}
	}