@Test public void testEncodeRequest() throws Exception { Request request = createRequest(); ChannelBuffer output = ChannelBuffers.dynamicBuffer(1024); codec.encode( channel, output, request ); byte[] bytes = new byte[output.readableBytes()]; output.readBytes(bytes); ByteArrayInputStream bis = new ByteArrayInputStream( bytes ); TTransport transport = new TIOStreamTransport( bis ); TBinaryProtocol protocol = new TBinaryProtocol( transport ); byte[] length = new byte[4]; transport.read( length, 0, 4 ); if ( bis.markSupported() ) { bis.mark( 0 ); } Assert.assertEquals( ThriftCodec.MAGIC, protocol.readI16() ); int messageLength = protocol.readI32(); Assert.assertEquals( messageLength + 4, bytes.length ); short headerLength = protocol.readI16(); Assert.assertEquals( ThriftCodec.VERSION, protocol.readByte() ); Assert.assertEquals( Demo.Iface.class.getName(), protocol.readString() ); Assert.assertEquals( request.getId(), protocol.readI64() ); if ( bis.markSupported() ) { bis.reset(); bis.skip( headerLength ); } TMessage message = protocol.readMessageBegin(); Demo.echoString_args args = new Demo.echoString_args(); args.read( protocol ); protocol.readMessageEnd(); Assert.assertEquals( "echoString", message.name ); Assert.assertEquals( TMessageType.CALL, message.type ); Assert.assertEquals( "Hello, World!", args.getArg() ); }
private void encodeRequest( Channel channel, ChannelBuffer buffer, Request request ) throws IOException { RpcInvocation inv = ( RpcInvocation ) request.getData(); int seqId = nextSeqId(); String serviceName = inv.getAttachment(Constants.INTERFACE_KEY); if ( StringUtils.isEmpty( serviceName ) ) { throw new IllegalArgumentException( new StringBuilder( 32 ) .append( "Could not find service name in attachment with key " ) .append(Constants.INTERFACE_KEY) .toString() ); } TMessage message = new TMessage( inv.getMethodName(), TMessageType.CALL, seqId ); String methodArgs = ExtensionLoader.getExtensionLoader( ClassNameGenerator.class ) .getExtension(channel.getUrl().getParameter(ThriftConstants.CLASS_NAME_GENERATOR_KEY, ThriftClassNameGenerator.NAME)) .generateArgsClassName(serviceName, inv.getMethodName()); if ( StringUtils.isEmpty( methodArgs ) ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, new StringBuilder(32).append( "Could not encode request, the specified interface may be incorrect." ).toString() ); } Class<?> clazz = cachedClass.get( methodArgs ); if ( clazz == null ) { try { clazz = ClassHelper.forNameWithThreadContextClassLoader( methodArgs ); cachedClass.putIfAbsent( methodArgs, clazz ); } catch ( ClassNotFoundException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } } TBase args; try { args = (TBase) clazz.newInstance(); } catch ( InstantiationException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } catch ( IllegalAccessException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } for( int i = 0; i < inv.getArguments().length; i++ ) { Object obj = inv.getArguments()[i]; if ( obj == null ) { continue; } TFieldIdEnum field = args.fieldForId( i + 1 ); String setMethodName = ThriftUtils.generateSetMethodName( field.getFieldName() ); Method method; try { method = clazz.getMethod( setMethodName, inv.getParameterTypes()[i] ); } catch ( NoSuchMethodException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } try { method.invoke( args, obj ); } catch ( IllegalAccessException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } catch ( InvocationTargetException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } } RandomAccessByteArrayOutputStream bos = new RandomAccessByteArrayOutputStream( 1024 ); TIOStreamTransport transport = new TIOStreamTransport( bos ); TBinaryProtocol protocol = new TBinaryProtocol( transport ); int headerLength, messageLength; byte[] bytes = new byte[4]; try { protocol.writeI16( MAGIC ); protocol.writeI32( Integer.MAX_VALUE ); protocol.writeI16( Short.MAX_VALUE ); protocol.writeByte( VERSION ); protocol.writeString( serviceName ); protocol.writeI64( request.getId() ); protocol.getTransport().flush(); headerLength = bos.size(); protocol.writeMessageBegin( message ); args.write( protocol ); protocol.writeMessageEnd(); protocol.getTransport().flush(); int oldIndex = messageLength = bos.size(); try { TFramedTransport.encodeFrameSize( messageLength, bytes ); bos.setWriteIndex( MESSAGE_LENGTH_INDEX ); protocol.writeI32( messageLength ); bos.setWriteIndex( MESSAGE_HEADER_LENGTH_INDEX ); protocol.writeI16( ( short )( 0xffff & headerLength ) ); } finally { bos.setWriteIndex( oldIndex ); } } catch ( TException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } buffer.writeBytes(bytes); buffer.writeBytes(bos.toByteArray()); }
ThriftCodec implements Codec2 { private void encodeRequest( Channel channel, ChannelBuffer buffer, Request request ) throws IOException { RpcInvocation inv = ( RpcInvocation ) request.getData(); int seqId = nextSeqId(); String serviceName = inv.getAttachment(Constants.INTERFACE_KEY); if ( StringUtils.isEmpty( serviceName ) ) { throw new IllegalArgumentException( new StringBuilder( 32 ) .append( "Could not find service name in attachment with key " ) .append(Constants.INTERFACE_KEY) .toString() ); } TMessage message = new TMessage( inv.getMethodName(), TMessageType.CALL, seqId ); String methodArgs = ExtensionLoader.getExtensionLoader( ClassNameGenerator.class ) .getExtension(channel.getUrl().getParameter(ThriftConstants.CLASS_NAME_GENERATOR_KEY, ThriftClassNameGenerator.NAME)) .generateArgsClassName(serviceName, inv.getMethodName()); if ( StringUtils.isEmpty( methodArgs ) ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, new StringBuilder(32).append( "Could not encode request, the specified interface may be incorrect." ).toString() ); } Class<?> clazz = cachedClass.get( methodArgs ); if ( clazz == null ) { try { clazz = ClassHelper.forNameWithThreadContextClassLoader( methodArgs ); cachedClass.putIfAbsent( methodArgs, clazz ); } catch ( ClassNotFoundException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } } TBase args; try { args = (TBase) clazz.newInstance(); } catch ( InstantiationException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } catch ( IllegalAccessException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } for( int i = 0; i < inv.getArguments().length; i++ ) { Object obj = inv.getArguments()[i]; if ( obj == null ) { continue; } TFieldIdEnum field = args.fieldForId( i + 1 ); String setMethodName = ThriftUtils.generateSetMethodName( field.getFieldName() ); Method method; try { method = clazz.getMethod( setMethodName, inv.getParameterTypes()[i] ); } catch ( NoSuchMethodException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } try { method.invoke( args, obj ); } catch ( IllegalAccessException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } catch ( InvocationTargetException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } } RandomAccessByteArrayOutputStream bos = new RandomAccessByteArrayOutputStream( 1024 ); TIOStreamTransport transport = new TIOStreamTransport( bos ); TBinaryProtocol protocol = new TBinaryProtocol( transport ); int headerLength, messageLength; byte[] bytes = new byte[4]; try { protocol.writeI16( MAGIC ); protocol.writeI32( Integer.MAX_VALUE ); protocol.writeI16( Short.MAX_VALUE ); protocol.writeByte( VERSION ); protocol.writeString( serviceName ); protocol.writeI64( request.getId() ); protocol.getTransport().flush(); headerLength = bos.size(); protocol.writeMessageBegin( message ); args.write( protocol ); protocol.writeMessageEnd(); protocol.getTransport().flush(); int oldIndex = messageLength = bos.size(); try { TFramedTransport.encodeFrameSize( messageLength, bytes ); bos.setWriteIndex( MESSAGE_LENGTH_INDEX ); protocol.writeI32( messageLength ); bos.setWriteIndex( MESSAGE_HEADER_LENGTH_INDEX ); protocol.writeI16( ( short )( 0xffff & headerLength ) ); } finally { bos.setWriteIndex( oldIndex ); } } catch ( TException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } buffer.writeBytes(bytes); buffer.writeBytes(bos.toByteArray()); } }
ThriftCodec implements Codec2 { private void encodeRequest( Channel channel, ChannelBuffer buffer, Request request ) throws IOException { RpcInvocation inv = ( RpcInvocation ) request.getData(); int seqId = nextSeqId(); String serviceName = inv.getAttachment(Constants.INTERFACE_KEY); if ( StringUtils.isEmpty( serviceName ) ) { throw new IllegalArgumentException( new StringBuilder( 32 ) .append( "Could not find service name in attachment with key " ) .append(Constants.INTERFACE_KEY) .toString() ); } TMessage message = new TMessage( inv.getMethodName(), TMessageType.CALL, seqId ); String methodArgs = ExtensionLoader.getExtensionLoader( ClassNameGenerator.class ) .getExtension(channel.getUrl().getParameter(ThriftConstants.CLASS_NAME_GENERATOR_KEY, ThriftClassNameGenerator.NAME)) .generateArgsClassName(serviceName, inv.getMethodName()); if ( StringUtils.isEmpty( methodArgs ) ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, new StringBuilder(32).append( "Could not encode request, the specified interface may be incorrect." ).toString() ); } Class<?> clazz = cachedClass.get( methodArgs ); if ( clazz == null ) { try { clazz = ClassHelper.forNameWithThreadContextClassLoader( methodArgs ); cachedClass.putIfAbsent( methodArgs, clazz ); } catch ( ClassNotFoundException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } } TBase args; try { args = (TBase) clazz.newInstance(); } catch ( InstantiationException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } catch ( IllegalAccessException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } for( int i = 0; i < inv.getArguments().length; i++ ) { Object obj = inv.getArguments()[i]; if ( obj == null ) { continue; } TFieldIdEnum field = args.fieldForId( i + 1 ); String setMethodName = ThriftUtils.generateSetMethodName( field.getFieldName() ); Method method; try { method = clazz.getMethod( setMethodName, inv.getParameterTypes()[i] ); } catch ( NoSuchMethodException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } try { method.invoke( args, obj ); } catch ( IllegalAccessException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } catch ( InvocationTargetException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } } RandomAccessByteArrayOutputStream bos = new RandomAccessByteArrayOutputStream( 1024 ); TIOStreamTransport transport = new TIOStreamTransport( bos ); TBinaryProtocol protocol = new TBinaryProtocol( transport ); int headerLength, messageLength; byte[] bytes = new byte[4]; try { protocol.writeI16( MAGIC ); protocol.writeI32( Integer.MAX_VALUE ); protocol.writeI16( Short.MAX_VALUE ); protocol.writeByte( VERSION ); protocol.writeString( serviceName ); protocol.writeI64( request.getId() ); protocol.getTransport().flush(); headerLength = bos.size(); protocol.writeMessageBegin( message ); args.write( protocol ); protocol.writeMessageEnd(); protocol.getTransport().flush(); int oldIndex = messageLength = bos.size(); try { TFramedTransport.encodeFrameSize( messageLength, bytes ); bos.setWriteIndex( MESSAGE_LENGTH_INDEX ); protocol.writeI32( messageLength ); bos.setWriteIndex( MESSAGE_HEADER_LENGTH_INDEX ); protocol.writeI16( ( short )( 0xffff & headerLength ) ); } finally { bos.setWriteIndex( oldIndex ); } } catch ( TException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } buffer.writeBytes(bytes); buffer.writeBytes(bos.toByteArray()); }  }
ThriftCodec implements Codec2 { private void encodeRequest( Channel channel, ChannelBuffer buffer, Request request ) throws IOException { RpcInvocation inv = ( RpcInvocation ) request.getData(); int seqId = nextSeqId(); String serviceName = inv.getAttachment(Constants.INTERFACE_KEY); if ( StringUtils.isEmpty( serviceName ) ) { throw new IllegalArgumentException( new StringBuilder( 32 ) .append( "Could not find service name in attachment with key " ) .append(Constants.INTERFACE_KEY) .toString() ); } TMessage message = new TMessage( inv.getMethodName(), TMessageType.CALL, seqId ); String methodArgs = ExtensionLoader.getExtensionLoader( ClassNameGenerator.class ) .getExtension(channel.getUrl().getParameter(ThriftConstants.CLASS_NAME_GENERATOR_KEY, ThriftClassNameGenerator.NAME)) .generateArgsClassName(serviceName, inv.getMethodName()); if ( StringUtils.isEmpty( methodArgs ) ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, new StringBuilder(32).append( "Could not encode request, the specified interface may be incorrect." ).toString() ); } Class<?> clazz = cachedClass.get( methodArgs ); if ( clazz == null ) { try { clazz = ClassHelper.forNameWithThreadContextClassLoader( methodArgs ); cachedClass.putIfAbsent( methodArgs, clazz ); } catch ( ClassNotFoundException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } } TBase args; try { args = (TBase) clazz.newInstance(); } catch ( InstantiationException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } catch ( IllegalAccessException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } for( int i = 0; i < inv.getArguments().length; i++ ) { Object obj = inv.getArguments()[i]; if ( obj == null ) { continue; } TFieldIdEnum field = args.fieldForId( i + 1 ); String setMethodName = ThriftUtils.generateSetMethodName( field.getFieldName() ); Method method; try { method = clazz.getMethod( setMethodName, inv.getParameterTypes()[i] ); } catch ( NoSuchMethodException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } try { method.invoke( args, obj ); } catch ( IllegalAccessException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } catch ( InvocationTargetException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } } RandomAccessByteArrayOutputStream bos = new RandomAccessByteArrayOutputStream( 1024 ); TIOStreamTransport transport = new TIOStreamTransport( bos ); TBinaryProtocol protocol = new TBinaryProtocol( transport ); int headerLength, messageLength; byte[] bytes = new byte[4]; try { protocol.writeI16( MAGIC ); protocol.writeI32( Integer.MAX_VALUE ); protocol.writeI16( Short.MAX_VALUE ); protocol.writeByte( VERSION ); protocol.writeString( serviceName ); protocol.writeI64( request.getId() ); protocol.getTransport().flush(); headerLength = bos.size(); protocol.writeMessageBegin( message ); args.write( protocol ); protocol.writeMessageEnd(); protocol.getTransport().flush(); int oldIndex = messageLength = bos.size(); try { TFramedTransport.encodeFrameSize( messageLength, bytes ); bos.setWriteIndex( MESSAGE_LENGTH_INDEX ); protocol.writeI32( messageLength ); bos.setWriteIndex( MESSAGE_HEADER_LENGTH_INDEX ); protocol.writeI16( ( short )( 0xffff & headerLength ) ); } finally { bos.setWriteIndex( oldIndex ); } } catch ( TException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } buffer.writeBytes(bytes); buffer.writeBytes(bos.toByteArray()); }  void encode( Channel channel, ChannelBuffer buffer, Object message ); Object decode( Channel channel, ChannelBuffer buffer ); }
ThriftCodec implements Codec2 { private void encodeRequest( Channel channel, ChannelBuffer buffer, Request request ) throws IOException { RpcInvocation inv = ( RpcInvocation ) request.getData(); int seqId = nextSeqId(); String serviceName = inv.getAttachment(Constants.INTERFACE_KEY); if ( StringUtils.isEmpty( serviceName ) ) { throw new IllegalArgumentException( new StringBuilder( 32 ) .append( "Could not find service name in attachment with key " ) .append(Constants.INTERFACE_KEY) .toString() ); } TMessage message = new TMessage( inv.getMethodName(), TMessageType.CALL, seqId ); String methodArgs = ExtensionLoader.getExtensionLoader( ClassNameGenerator.class ) .getExtension(channel.getUrl().getParameter(ThriftConstants.CLASS_NAME_GENERATOR_KEY, ThriftClassNameGenerator.NAME)) .generateArgsClassName(serviceName, inv.getMethodName()); if ( StringUtils.isEmpty( methodArgs ) ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, new StringBuilder(32).append( "Could not encode request, the specified interface may be incorrect." ).toString() ); } Class<?> clazz = cachedClass.get( methodArgs ); if ( clazz == null ) { try { clazz = ClassHelper.forNameWithThreadContextClassLoader( methodArgs ); cachedClass.putIfAbsent( methodArgs, clazz ); } catch ( ClassNotFoundException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } } TBase args; try { args = (TBase) clazz.newInstance(); } catch ( InstantiationException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } catch ( IllegalAccessException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } for( int i = 0; i < inv.getArguments().length; i++ ) { Object obj = inv.getArguments()[i]; if ( obj == null ) { continue; } TFieldIdEnum field = args.fieldForId( i + 1 ); String setMethodName = ThriftUtils.generateSetMethodName( field.getFieldName() ); Method method; try { method = clazz.getMethod( setMethodName, inv.getParameterTypes()[i] ); } catch ( NoSuchMethodException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } try { method.invoke( args, obj ); } catch ( IllegalAccessException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } catch ( InvocationTargetException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } } RandomAccessByteArrayOutputStream bos = new RandomAccessByteArrayOutputStream( 1024 ); TIOStreamTransport transport = new TIOStreamTransport( bos ); TBinaryProtocol protocol = new TBinaryProtocol( transport ); int headerLength, messageLength; byte[] bytes = new byte[4]; try { protocol.writeI16( MAGIC ); protocol.writeI32( Integer.MAX_VALUE ); protocol.writeI16( Short.MAX_VALUE ); protocol.writeByte( VERSION ); protocol.writeString( serviceName ); protocol.writeI64( request.getId() ); protocol.getTransport().flush(); headerLength = bos.size(); protocol.writeMessageBegin( message ); args.write( protocol ); protocol.writeMessageEnd(); protocol.getTransport().flush(); int oldIndex = messageLength = bos.size(); try { TFramedTransport.encodeFrameSize( messageLength, bytes ); bos.setWriteIndex( MESSAGE_LENGTH_INDEX ); protocol.writeI32( messageLength ); bos.setWriteIndex( MESSAGE_HEADER_LENGTH_INDEX ); protocol.writeI16( ( short )( 0xffff & headerLength ) ); } finally { bos.setWriteIndex( oldIndex ); } } catch ( TException e ) { throw new RpcException( RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e ); } buffer.writeBytes(bytes); buffer.writeBytes(bos.toByteArray()); }  void encode( Channel channel, ChannelBuffer buffer, Object message ); Object decode( Channel channel, ChannelBuffer buffer ); static final int MESSAGE_LENGTH_INDEX; static final int MESSAGE_HEADER_LENGTH_INDEX; static final int MESSAGE_SHORTEST_LENGTH; static final String NAME; static final String PARAMETER_CLASS_NAME_GENERATOR; static final byte VERSION; static final short MAGIC; }