@Override
public Object copy(@Nonnull ClassLoader targetClassLoader, @Nonnull Object what) {
  Objects.requireNonNull(targetClassLoader);
  if (!(what instanceof Message)) {
    throw new IllegalStateException();
  }
  Message message = (Message) what;
  ByteString messageBytes = message.toByteString();
  try {
    Parser<? extends Message> parser =
        parserForClassName(targetClassLoader, what.getClass().getName());
    return parser.parseFrom(messageBytes);
  } catch (InvalidProtocolBufferException | ClassNotFoundException e) {
    throw new IllegalStateException(e);
  }
}
 

public Subscriber() {
    Parser<T> parser = null;
    try {
        Class<?> cl = getClass();
        while (!Subscriber.class.equals(cl.getSuperclass())) {
            // case of multiple inheritance, we are trying to get the
            // first available generic info
            if (cl.getGenericSuperclass() instanceof ParameterizedType) {
                break;
            }
            cl = cl.getSuperclass();
        }
        Class<T> type = ((Class<T>) ((ParameterizedType) cl.getGenericSuperclass())
                .getActualTypeArguments()[0]);
        parser = (Parser<T>) type.getDeclaredField("PARSER").get(null);
    } catch (Exception e) {
        System.out.println("Error: callback creation failed");
        e.printStackTrace();
    }
    this.parser = parser;
}
 

/**
 * Decode a ByteBuf body from protobuf format to an object of designated Class type.
 *
 * @param body current http request
 * @param clazz target class for decoding
 * @return object of type clazz
 */
@Override
@SuppressWarnings("unchecked")
public <T> T decode(ByteBuf body, CharSequence contentType, Class<T> clazz) throws IOException {
  // TODO (AD): given a Content-Type of application/protobuf; proto=org.some.Message,
  // we currently ignore the 2nd part, but should at least validate it in the future.

  if (!MessageLite.class.isAssignableFrom(clazz)) {
    throw new IllegalArgumentException(
        String.format("%s does not extend from MessageLite", clazz.getName()));
  }

  MessageLite message = protoDefaultInstances.get(clazz);
  Parser<?> parser = message.getParserForType();
  try (ByteBufInputStream stream = new ByteBufInputStream(body)) {
    return (T) parser.parseFrom(stream);
  }
}
 

@SuppressWarnings("unchecked")
@Override
public <P extends com.google.protobuf.Message> Serializer<P, byte[]> visitProtobufFormat(Class<P> clazz) {

  Parser<P> parser;
  try {
    Method defaultInstanceGetter = clazz.getDeclaredMethod("getDefaultInstance");
    com.google.protobuf.Message defaultInst = (com.google.protobuf.Message) defaultInstanceGetter.invoke(null);
    parser = (Parser<P>) defaultInst.getParserForType();
    Preconditions.checkNotNull(parser);
  } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
    throw new DatastoreFatalException("Unable to get the parser for " + clazz.getName(), e);
  }

  return new ProtobufSerializer<>(clazz, parser);
}
 

@SuppressWarnings("unchecked")
@Override
public XMessageHeader readHeader() throws IOException {
    if (!this.hasReadHeader) {
        try {
            readMessageHeader();
        } catch (IOException ex) {
            throw new CJCommunicationsException("Cannot read packet header", ex);
        }
    }
    int type = this.header.getMessageType(); // forces header read if necessary

    Class<? extends GeneratedMessage> messageClass = MessageConstants.getMessageClassForType(type);

    if (messageClass == Error.class) {
        // throw an error/exception if receive an Error message
        throw new XProtocolError(readAndParse((Parser<Error>) MessageConstants.MESSAGE_CLASS_TO_PARSER.get(Error.class)));
    }

    return this.header;
}
 

/** Get the memoized {@link Parser}, possibly initializing it lazily. */
protected Parser<T> getParser() {
  if (memoizedParser == null) {
    try {
      if (DynamicMessage.class.equals(protoMessageClass)) {
        throw new IllegalArgumentException(
            "DynamicMessage is not supported by the ProtoCoder, use the DynamicProtoCoder.");
      } else {
        @SuppressWarnings("unchecked")
        T protoMessageInstance =
            (T) protoMessageClass.getMethod("getDefaultInstance").invoke(null);
        @SuppressWarnings("unchecked")
        Parser<T> tParser = (Parser<T>) protoMessageInstance.getParserForType();
        memoizedParser = tParser;
      }
    } catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException e) {
      throw new IllegalArgumentException(e);
    }
  }
  return memoizedParser;
}
 

private <T> T parseAny(Any value, Parser<T> parser) {
  try {
    return parser.parseFrom(value.getValue());
  } catch (InvalidProtocolBufferException e) {
    throw new RuntimeException(e);
  }
}
 

public Subscriber(Class<T> type) {
    Parser<T> parser = null;
    try {
        parser = (Parser<T>) type.getDeclaredField("PARSER").get(null);
    } catch (Exception e) {
        System.out.println("Error: callback creation failed");
        e.printStackTrace();
    }
    this.parser = parser;
}
 

public static <T> T get(ByteBuf pBody, Parser<T> parser) throws RpcException {
  try {
    ByteBufInputStream is = new ByteBufInputStream(pBody);
    return parser.parseFrom(is);
  } catch (InvalidProtocolBufferException e) {
    throw new RpcException(
        String.format("Failure while decoding message with parser of type. %s",
            parser.getClass().getCanonicalName()), e);
  }
}
 

@SuppressWarnings("unchecked")
@Override
public <REQUEST extends MessageLite, RESPONSE extends MessageLite> void registerCustomHandler(int messageTypeId,
    CustomMessageHandler<REQUEST, RESPONSE> handler, Parser<REQUEST> parser) {
  handlerRegistry.registerCustomHandler(
      messageTypeId,
      handler,
      new ControlTunnel.ProtoSerDe<REQUEST>(parser),
      (CustomSerDe<RESPONSE>) new ControlTunnel.ProtoSerDe<Message>(null));
}
 

@Override
protected <T> ListenableFuture<T> expect(
    Digest digest, Parser<T> parser, Executor executor, RequestMetadata requestMetadata) {
  Context.CancellableContext withDeadline =
      Context.current().withDeadlineAfter(60, SECONDS, contextDeadlineScheduler);
  Context previousContext = withDeadline.attach();
  try {
    ListenableFuture<T> future = super.expect(digest, parser, executor, requestMetadata);
    future.addListener(() -> withDeadline.cancel(null), directExecutor());
    return future;
  } finally {
    withDeadline.detach(previousContext);
  }
}
 

private static void equalsTest(p_test.p_testMsg msg) throws IOException {
    final Parser<p_test.p_testMsg> parser = msg.getParserForType();

    final CodedOutputStream codedOutputStream = CodedOutputStream.newInstance(buffer);
    writeTypeId(msg, codedOutputStream);
    msg.writeTo(codedOutputStream);
    System.out.println("encode result bytes = " + codedOutputStream.getTotalBytesWritten());

    final CodedInputStream inputStream = CodedInputStream.newInstance(buffer, 0, codedOutputStream.getTotalBytesWritten());
    final Class<?> type = readType(inputStream);
    final Object decodeMsg = parser.parseFrom(inputStream);
    System.out.println("codec equals result = " + msg.equals(decodeMsg));
}
 

private static void codecTest(Message msg, int loopTimes, Map<Class<?>, Parser<? extends Message>> parserMap) throws IOException {
    final long start = System.currentTimeMillis();
    for (int index = 0; index < loopTimes; index++) {
        // 这里需要简单模拟下解码过程
        final CodedOutputStream codedOutputStream = CodedOutputStream.newInstance(buffer);
        writeTypeId(msg, codedOutputStream);
        msg.writeTo(codedOutputStream);

        final CodedInputStream inputStream = CodedInputStream.newInstance(buffer, 0, codedOutputStream.getTotalBytesWritten());
        final Class<?> messageClass = readType(inputStream);
        final Parser<?> parser = parserMap.get(messageClass);
        final Object decodeMsg = parser.parseFrom(inputStream);
    }
    System.out.println("codec " + loopTimes + " times cost timeMs " + (System.currentTimeMillis() - start));
}
 

/**
 * 寻找protoBuf消息的parser对象
 *
 * @param clazz protoBuffer class
 * @return parser
 */
public static <T extends Message> Parser<T> findParser(@Nonnull Class<T> clazz) {
    Objects.requireNonNull(clazz);
    try {
        final Method method = clazz.getDeclaredMethod("getDefaultInstance");
        method.setAccessible(true);
        final Message instance = (Message) method.invoke(null);
        @SuppressWarnings("unchecked") final Parser<T> parserForType = (Parser<T>) instance.getParserForType();
        return parserForType;
    } catch (Exception e) {
        throw new IllegalArgumentException("bad class " + clazz.getName(), e);
    }
}
 

@Override
public <T> T readMessage(Parser<T> parser) throws Exception {
    final BinaryTag tag = inputStream.readTag();
    if (tag == BinaryTag.NULL) {
        return null;
    }
    checkTag(tag, BinaryTag.POJO);

    @SuppressWarnings("unchecked") final T message = (T) PojoCodec.readPojoImp(inputStream, codecRegistry);
    return message;
}
 

/**
 * @param typeModelMapper 类型映射信息
 * @param filter          由于{@link BinarySerializer}支持的消息类是确定的，不能加入，但是允许过滤删除
 */
@SuppressWarnings("unchecked")
public static BinarySerializer newInstance(TypeModelMapper typeModelMapper, Predicate<Class<?>> filter) {
    final Set<Class<?>> supportedClassSet = getFilteredSupportedClasses(filter);
    final List<PojoCodecImpl<?>> codecList = new ArrayList<>(supportedClassSet.size());
    try {
        for (Class<?> messageClazz : supportedClassSet) {
            // protoBuf消息
            if (Message.class.isAssignableFrom(messageClazz)) {
                Parser<?> parser = ProtoUtils.findParser((Class<? extends Message>) messageClazz);
                codecList.add(new ProtoMessageCodec(messageClazz, parser));
                continue;
            }

            // protoBufEnum
            if (ProtocolMessageEnum.class.isAssignableFrom(messageClazz)) {
                final Internal.EnumLiteMap<?> mapper = ProtoUtils.findMapper((Class<? extends ProtocolMessageEnum>) messageClazz);
                codecList.add(new ProtoEnumCodec(messageClazz, mapper));
                continue;
            }

            // 带有DBEntity和SerializableClass注解的所有类，和手写Serializer的类
            final Class<? extends PojoCodecImpl<?>> serializerClass = CodecScanner.getCodecClass(messageClazz);
            if (serializerClass != null) {
                final PojoCodecImpl<?> codec = createCodecInstance(serializerClass);
                codecList.add(new CustomPojoCodec(codec));
                continue;
            }

            throw new IllegalArgumentException("Unsupported class " + messageClazz.getName());
        }

        final CodecRegistry codecRegistry = CodecRegistrys.fromAppPojoCodecs(typeModelMapper, codecList);
        return new BinarySerializer(codecRegistry);
    } catch (Exception e) {
        return ExceptionUtils.rethrow(e);
    }
}
 

@Override
public Object deserialize(@Nonnull ClassLoader targetClassLoader, @Nonnull Payload payload) {
  try {
    Parser<? extends Message> parser =
        parserForClassName(targetClassLoader, payload.getClassName());
    return parser.parseFrom(payload.getPayloadBytes());
  } catch (InvalidProtocolBufferException | ClassNotFoundException e) {
    throw new IllegalStateException(e);
  }
}
 

private Parser<? extends Message> parserForClassName(
    ClassLoader userCodeClassLoader, String messageClassName) throws ClassNotFoundException {

  ObjectOpenHashMap<ClassLoader, Parser<? extends Message>> classLoaders =
      PARSER_CACHE.get(messageClassName);
  if (classLoaders == null) {
    PARSER_CACHE.put(messageClassName, classLoaders = new ObjectOpenHashMap<>());
  }
  Parser<? extends Message> parser = classLoaders.get(userCodeClassLoader);
  if (parser == null) {
    classLoaders.put(
        userCodeClassLoader, parser = findParser(userCodeClassLoader, messageClassName));
  }
  return parser;
}
 

private Parser<? extends Message> findParser(
    ClassLoader userCodeClassLoader, String messageClassName) throws ClassNotFoundException {
  Class<? extends Message> messageType =
      Class.forName(messageClassName, true, userCodeClassLoader).asSubclass(Message.class);

  return ProtobufReflectionUtil.protobufParser(messageType);
}
 

/**
 * Generic method to convert Protostuff to Protobuf. Uses LegacyProtobufSerializer because deserializing with the
 * regular protobuf serializer does not handle repeated fields correctly.
 * @param protobufParser Parser for protobuf object
 * @param protostuff Protostuff object to convert
 * @param <M> Type of Protobuf
 * @param <T> Type of Protobuff
 * @return Converted object as Protobuf
 */
private static <M extends GeneratedMessageV3, T extends Message<T> & Schema<T>>
M toBuf(Parser<M> protobufParser, T protostuff) {
  try {
    LinkedBuffer buffer = LinkedBuffer.allocate();
    byte[] bytes = ProtobufIOUtil.toByteArray(protostuff, protostuff.cachedSchema(), buffer);
    // LegacyProtobufSerializer is necessary as it deals with stuff/buf grouping differences
    return LegacyProtobufSerializer.parseFrom(protobufParser, bytes);
  } catch (InvalidProtocolBufferException e) {
    throw new IllegalArgumentException("Cannot convert from protostuff to protobuf");
  }
}
 

@SuppressWarnings("unchecked")
public static <M extends Message> Parser<M> protobufParser(Class<M> messageClass) {
  Object parser = getParserFromGeneratedMessage(messageClass);
  if (!(parser instanceof Parser)) {
    throw new IllegalStateException(
        "was expecting a protobuf parser to be return from the static parser() method on the type  "
            + messageClass
            + " but instead got "
            + parser);
  }
  return (Parser<M>) parser;
}
 

@SuppressWarnings("unchecked")
public static <M extends Message> Parser<M> protobufParser(Class<M> messageClass) {
  Object parser = getParserFromGeneratedMessage(messageClass);
  if (!(parser instanceof Parser)) {
    throw new IllegalStateException(
        "was expecting a protobuf parser to be return from the static parser() method on the type  "
            + messageClass
            + " but instead got "
            + parser);
  }
  return (Parser<M>) parser;
}
 

@Override
public Object deserialize(@Nonnull ClassLoader targetClassLoader, @Nonnull Payload payload) {
  try {
    Parser<? extends Message> parser =
        parserForClassName(targetClassLoader, payload.getClassName());
    return parser.parseFrom(payload.getPayloadBytes());
  } catch (InvalidProtocolBufferException | ClassNotFoundException e) {
    throw new IllegalStateException(e);
  }
}
 

private Parser<? extends Message> parserForClassName(
    ClassLoader userCodeClassLoader, String messageClassName) throws ClassNotFoundException {

  ObjectOpenHashMap<ClassLoader, Parser<? extends Message>> classLoaders =
      PARSER_CACHE.get(messageClassName);
  if (classLoaders == null) {
    PARSER_CACHE.put(messageClassName, classLoaders = new ObjectOpenHashMap<>());
  }
  Parser<? extends Message> parser = classLoaders.get(userCodeClassLoader);
  if (parser == null) {
    classLoaders.put(
        userCodeClassLoader, parser = findParser(userCodeClassLoader, messageClassName));
  }
  return parser;
}
 

private Parser<? extends Message> findParser(
    ClassLoader userCodeClassLoader, String messageClassName) throws ClassNotFoundException {
  Class<? extends Message> messageType =
      Class.forName(messageClassName, true, userCodeClassLoader).asSubclass(Message.class);

  return ProtobufReflectionUtil.protobufParser(messageType);
}
 

public static <M extends Message> M parseProtobufOrThrow(Parser<M> parser, InputStream input) {
  try {
    return parser.parseFrom(input);
  } catch (IOException e) {
    throw new IllegalStateException("Unable to parse a Protobuf message", e);
  }
}
 

ProtoBufSerializationProvider(final Class<T> targetClass, final GrpcMessageEncoding messageEncoding,
                              final Parser<T> parser) {
    this.targetClass = targetClass;
    this.messageEncoding = messageEncoding;
    this.serializer = new ProtoSerializer(messageEncoding);
    this.parser = parser;
}
 

@Override
public <X> StreamingDeserializer<X> getDeserializer(final Class<X> classToDeSerialize) {
    if (targetClass != classToDeSerialize) {
        throw new SerializationException("Unknown class to deserialize: " + classToDeSerialize.getName());
    }
    @SuppressWarnings("unchecked")
    Parser<X> parser = (Parser<X>) this.parser;
    return new ProtoDeserializer<>(parser, messageEncoding);
}
 

@Override
@SuppressWarnings("unchecked")
public <T> StreamingDeserializer<T> getDeserializer(final Class<T> classToDeSerialize) {
    requireMessageLite(classToDeSerialize);
    Parser<T> parser = parsers.computeIfAbsent(classToDeSerialize, parserForClass::apply);
    return new ProtobufDeserializer<>(parser);
}
 

@SuppressWarnings("unchecked")
private <T extends GeneratedMessage> T parseNotice(ByteString payload, Class<T> noticeClass) {
    try {
        Parser<T> parser = (Parser<T>) MessageConstants.MESSAGE_CLASS_TO_PARSER.get(noticeClass);
        return parser.parseFrom(payload);
    } catch (InvalidProtocolBufferException ex) {
        throw new CJCommunicationsException(ex);
    }
}