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(INTERFACE_KEY); if (StringUtils.isEmpty(serviceName)) { throw new IllegalArgumentException("Could not find service name in attachment with key " + INTERFACE_KEY); } 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, "Could not encode request, the specified interface may be incorrect."); } Class<?> clazz = CACHED_CLASS.get(methodArgs); if (clazz == null) { try { clazz = ClassUtils.forNameWithThreadContextClassLoader(methodArgs); CACHED_CLASS.putIfAbsent(methodArgs, clazz); } catch (ClassNotFoundException e) { throw new RpcException(RpcException.SERIALIZATION_EXCEPTION, e.getMessage(), e); } } TBase args; try { args = (TBase) clazz.newInstance(); } catch (InstantiationException | 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 | 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.writeString(inv.getAttachment(PATH_KEY)); 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()); }  @Override void encode(Channel channel, ChannelBuffer buffer, Object message); @Override 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; }
@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); } Assertions.assertEquals(ThriftCodec.MAGIC, protocol.readI16()); int messageLength = protocol.readI32(); Assertions.assertEquals(messageLength + 4, bytes.length); short headerLength = protocol.readI16(); Assertions.assertEquals(ThriftCodec.VERSION, protocol.readByte()); Assertions.assertEquals(Demo.Iface.class.getName(), protocol.readString()); Assertions.assertEquals(Demo.Iface.class.getName(), protocol.readString()); Assertions.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(); Assertions.assertEquals("echoString", message.name); Assertions.assertEquals(TMessageType.CALL, message.type); Assertions.assertEquals("Hello, World!", args.getArg()); }