Hessian远程调用（基于Netty）

 

public class HessianMethodInvocation implements InvocationHandler, Serializable {

	private static final long serialVersionUID = 7304512179586775133L;
	
	private TestServiceImpl testService;
	private HessianSkeleton skeleton;
	
	private AbstractHessianInput input;
	private AbstractHessianOutput output;
	
	public HessianMethodInvocation(AbstractHessianInput input, AbstractHessianOutput output) {
		testService = new TestServiceImpl();
		skeleton = new HessianSkeleton(testService, TestService.class);
		
		this.input = input;
		this.output = output;
	}
	
	public void invoke() {
		try {
			skeleton.invoke(input, output);
		} catch (Throwable e) {
			e.printStackTrace();
		}
	}
	
	@Override
	public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
		return null;
	}

}

 

 

public class HessianDecoder extends ByteToMessageDecoder {

	private final ClassResolver classResolver;
	
	private static TestServiceImpl testService;
	
	private static HessianSkeleton skeleton;
	
	static {
		testService = new TestServiceImpl();
		
		skeleton = new HessianSkeleton(testService, TestService.class);
	}
	
	public HessianDecoder(ClassResolver classResolver) {
        this(1048576, classResolver);
    }
	
	public HessianDecoder(int maxObjectSize, ClassResolver classResolver) {
        super();
        this.classResolver = classResolver;
    }

	@Override
	protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
		int length = in.readInt();
		
		ByteBuf frame = Unpooled.buffer(length);
		in.readBytes(frame);
        if (frame == null) {
            return;
        }
        
        ByteArrayInputStream is = new ByteArrayInputStream(frame.array());
		
		
		OutputStream osToUse = new ByteArrayOutputStream();
		
		AbstractHessianInput input = new HessianInput(is);
		AbstractHessianOutput output = new HessianOutput(osToUse);
		
		HessianMethodInvocation invocation = new HessianMethodInvocation(input, output);
        if (invocation != null) {
            out.add(invocation);
        }
	}
}

 

 

            public void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline p = ch.pipeline();
//                if (sslCtx != null) {
//                    p.addLast(sslCtx.newHandler(ch.alloc()));
//                }
//                p.addLast(new LoggingHandler(LogLevel.INFO));
                p.addLast(
                    new HessianDecoder(ClassResolvers.cacheDisabled(null)),
                    new HessianMethodInvocationHandler());
            }

 

 

class HessianMethodInvocationHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
    	System.out.println(msg.getClass().getName());
    	if (! (msg instanceof HessianMethodInvocation)) {
    		
    	}
    	HessianMethodInvocation invocation = (HessianMethodInvocation) msg;
    	invocation.invoke();
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // Close the connection when an exception is raised.
        cause.printStackTrace();
        ctx.close();
    }
}

 

 

public class HessianMethodInvocationProxy implements InvocationHandler, Serializable {

	private static final long serialVersionUID = 6587104138220238229L;
	
	private String method;
	private Object[] args;
	
	public HessianMethodInvocationProxy(String method, Object[] args) {
		this.method = method;
		this.args = args;
	}
	
	public String getMethod() {
		return method;
	}

	public void setMethod(String method) {
		this.method = method;
	}

	public Object[] getArgs() {
		return args;
	}

	public void setArgs(Object[] args) {
		this.args = args;
	}

	public void invoke(ChannelHandlerContext ctx) {
		ctx.writeAndFlush(this);
	}
	
	@Override
	public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
		return null;
	}

}

 

 

public class HessianEncoder extends MessageToByteEncoder<Serializable> {

	private static final byte[] LENGTH_PLACEHOLDER = new byte[4];
	
	@Override
	protected void encode(ChannelHandlerContext ctx, Serializable msg, ByteBuf out) throws Exception {
		if (! (msg instanceof HessianMethodInvocationProxy)) {
			
		}
		HessianMethodInvocationProxy proxy = (HessianMethodInvocationProxy) msg;
		int startIdx = out.writerIndex();

        ByteBufOutputStream bout = new ByteBufOutputStream(out);
//        ObjectOutputStream oout = null;
        try {
            bout.write(LENGTH_PLACEHOLDER);
            
            ByteArrayOutputStream osToUse = new ByteArrayOutputStream();
			AbstractHessianOutput output = new HessianOutput(osToUse);
			try {
				output.call(proxy.getMethod(), proxy.getArgs());
			} catch (Throwable e) {
				e.printStackTrace();
			}
			bout.write(osToUse.toByteArray());
        } finally {
                bout.close();
        }

        int endIdx = out.writerIndex();

        out.setInt(startIdx, endIdx - startIdx - 4);
	}

}

 

 

				public void initChannel(SocketChannel ch) throws Exception {
					ChannelPipeline p = ch.pipeline();
//					if (sslCtx != null) {
//						p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
//					}
					p.addLast(new HessianEncoder());
					p.addLast(new HessianMethodInvocationHandlerProxy());
				}

 

 

class HessianMethodInvocationHandlerProxy extends ChannelInboundHandlerAdapter {

	/**
	 * Creates a client-side handler.
	 */
	public HessianMethodInvocationHandlerProxy() {
		//TODO
	}


	@Override
	public void channelActive(ChannelHandlerContext ctx) {
		HessianMethodInvocationProxy proxy = new HessianMethodInvocationProxy("test", new Object[] {"helloworld1234567"});
		proxy.invoke(ctx);
	}


	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) {
		System.out.println(msg);
	}


	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) {
		ctx.flush();
	}


	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
		// Close the connection when an exception is raised.
		cause.printStackTrace();
		ctx.close();
	}
}