<properties>
    <grpc.version>1.42.1</grpc.version>
  </properties>

<dependencies>
    <dependency>
      <groupId>javax.annotation</groupId>
      <artifactId>javax.annotation-api</artifactId>
      <version>1.3.2</version>
      <scope>provided</scope>
    </dependency>
    <dependency>
      <groupId>io.grpc</groupId>
      <artifactId>grpc-netty-shaded</artifactId>
      <version>${grpc.version}</version>
      <scope>runtime</scope>
    </dependency>
    <dependency>
      <groupId>io.grpc</groupId>
      <artifactId>grpc-protobuf</artifactId>
      <version>${grpc.version}</version>
    </dependency>
    <dependency>
      <groupId>io.grpc</groupId>
      <artifactId>grpc-stub</artifactId>
      <version>${grpc.version}</version>
    </dependency>
    <dependency>
      <groupId>io.grpc</groupId>
      <artifactId>grpc-testing</artifactId>
      <version>${grpc.version}</version>
      <scope>test</scope>
    </dependency>
  </dependencies>

  <properties>
    <protobuf.output.directory>${project.build.directory}/generated-sources</protobuf.output.directory>
    <protobuf.input.directory>${project.build.directory}/proto</protobuf.input.directory>
  </properties>

<plugin>
        <groupId>com.googlecode.maven-download-plugin</groupId>
        <artifactId>download-maven-plugin</artifactId>
        <version>1.6.0</version>
        <executions>
          <execution>
            <id>getCommonProto</id>
            <!-- the wget goal actually binds itself to this phase by default -->
            <phase>initialize</phase>
            <goals>
              <goal>wget</goal>
            </goals>
            <configuration>
              <url>${dapr.proto.baseurl}/common/v1/common.proto</url>
              <outputFileName>common.proto</outputFileName>
              <!-- default target location, just to demonstrate the parameter -->
              <outputDirectory>${protobuf.input.directory}/dapr/proto/common/v1</outputDirectory>
            </configuration>
          </execution>
          <execution>
            <id>getDaprProto</id>
            <!-- the wget goal actually binds itself to this phase by default -->
            <phase>initialize</phase>
            <goals>
              <goal>wget</goal>
            </goals>
            <configuration>
              <url>${dapr.proto.baseurl}/runtime/v1/dapr.proto</url>
              <outputFileName>dapr.proto</outputFileName>
              <!-- default target location, just to demonstrate the parameter -->
              <outputDirectory>${protobuf.input.directory}</outputDirectory>
            </configuration>
          </execution>
          <execution>
            <id>getDaprClientProto</id>
            <!-- the wget goal actually binds itself to this phase by default -->
            <phase>initialize</phase>
            <goals>
              <goal>wget</goal>
            </goals>
            <configuration>
              <url>${dapr.proto.baseurl}/runtime/v1/appcallback.proto</url>
              <outputFileName>appcallback.proto</outputFileName>
              <!-- default target location, just to demonstrate the parameter -->
              <outputDirectory>${protobuf.input.directory}</outputDirectory>
            </configuration>
          </execution>
        </executions>
      </plugin>

<plugin>
        <groupId>com.github.os72</groupId>
        <artifactId>protoc-jar-maven-plugin</artifactId>
        <version>3.11.4</version>
        <executions>
          <execution>
            <phase>generate-sources</phase>
            <goals>
              <goal>run</goal>
            </goals>
            <configuration>
              <protocVersion>${protobuf.version}</protocVersion>
              <addProtoSources>inputs</addProtoSources>
              <includeMavenTypes>direct</includeMavenTypes>
              <includeStdTypes>true</includeStdTypes>
              <inputDirectories>
                <include>${protobuf.input.directory}/dapr/proto/common/v1</include>
                <include>${protobuf.input.directory}</include>
              </inputDirectories>
              <outputTargets>
                <outputTarget>
                  <type>java</type>
                  <outputDirectory>${protobuf.output.directory}</outputDirectory>
                </outputTarget>
                <outputTarget>
                  <type>grpc-java</type>
                  <outputDirectory>${protobuf.output.directory}</outputDirectory>
                  <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}</pluginArtifact>
                </outputTarget>
              </outputTargets>
            </configuration>
          </execution>
        </executions>
      </plugin>

      <plugin>
        <groupId>com.github.spotbugs</groupId>
        <artifactId>spotbugs-maven-plugin</artifactId>
        <configuration>
          <!-- Skip findbugs for auto-generated code -->
          <skip>true</skip>
        </configuration>
      </plugin>

<plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-javadoc-plugin</artifactId>
    <version>3.2.0</version>
    <executions>
        <execution>
        <id>attach-javadocs</id>
        <goals>
            <goal>jar</goal>
        </goals>
        </execution>
    </executions>
    </plugin>

<plugin>
        <groupId>org.apache.maven.plugins</groupId>
        <artifactId>maven-source-plugin</artifactId>
        <version>3.2.1</version>
        <executions>
          <execution>
            <id>attach-sources</id>
            <goals>
              <goal>jar-no-fork</goal>
            </goals>
          </execution>
        </executions>
      </plugin>

<dapr.proto.baseurl>https://raw.githubusercontent.com/dapr/dapr/v1.7.0-rc.2/dapr/proto</dapr.proto.baseurl>

// 对应用程序的对象进行序列化和反序列化
public interface DaprObjectSerializer {

  // 将给定的对象序列化为byte[].
  byte[] serialize(Object o) throws IOException;

  // 将给定的byte[]反序列化为一个对象。
  <T> T deserialize(byte[] data, TypeRef<T> type) throws IOException;

  // 返回请求的内容类型
  String getContentType();
}

public class DefaultObjectSerializer extends ObjectSerializer implements DaprObjectSerializer {

  @Override
  public byte[] serialize(Object o) throws IOException {
    return super.serialize(o);
  }

  @Override
  public <T> T deserialize(byte[] data, TypeRef<T> type) throws IOException {
    return super.deserialize(data, type);
  }

  @Override
  public String getContentType() {
    return "application/json";
  }
}

public class ObjectSerializer {
  // 默认构造函数，以避免类在包外被实例化，但仍可以被继承。
  protected ObjectSerializer() {
  }
}

  protected static final ObjectMapper OBJECT_MAPPER = new ObjectMapper()
      .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
      .setSerializationInclusion(JsonInclude.Include.NON_NULL);

public byte[] serialize(Object state) throws IOException {
    if (state == null) {
      return null;
    }

    if (state.getClass() == Void.class) {
      return null;
    }

    // Have this check here to be consistent with deserialization (see deserialize() method below).
    if (state instanceof byte[]) {
      return (byte[]) state;
    }

    // Proto buffer class is serialized directly.
    if (state instanceof MessageLite) {
      return ((MessageLite) state).toByteArray();
    }

    // Not string, not primitive, so it is a complex type: we use JSON for that.
    return OBJECT_MAPPER.writeValueAsBytes(state);
  }

  public <T> T deserialize(byte[] content, TypeRef<T> type) throws IOException {
    return deserialize(content, OBJECT_MAPPER.constructType(type.getType()));
  }

  public <T> T deserialize(byte[] content, Class<T> clazz) throws IOException {
    return deserialize(content, OBJECT_MAPPER.constructType(clazz));
  }

  private <T> T deserialize(byte[] content, JavaType javaType) throws IOException {
    // 对应 serialize 的做法
    if ((javaType == null) || javaType.isTypeOrSubTypeOf(Void.class)) {
      return null;
    }

    // 如果是 java 基本类型，则交给 deserializePrimitives() 方法处理
    // 注意此时 content 有可能是 null 或者 空数组
    if (javaType.isPrimitive()) {
      return deserializePrimitives(content, javaType);
    }

    // 对应 serialize 的做法
    if (content == null) {
      return null;
    }

    // Deserialization of GRPC response fails without this check since it does not come as base64 encoded byte[].
    // 如果没有这个检查，GRPC响应的反序列化就会失败，因为它不是以 base64 编码的 byte[] 形式出现的。
    // TBD：这里有点不是太理解
    if (javaType.hasRawClass(byte[].class)) {
      return (T) content;
    }

    // // 对应 serialize 的做法，但长度为零的检测放在 byte[] 检测之后
    if (content.length == 0) {
      return null;
    }

    // 对 CloudEvent 的支持：如果是 CloudEvent，则单独序列化
    if (javaType.hasRawClass(CloudEvent.class)) {
      return (T) CloudEvent.deserialize(content);
    }

    // 对 grpc MessageLite 的支持：通过反射调用 parseFrom 方法
    if (javaType.isTypeOrSubTypeOf(MessageLite.class)) {
      try {
        Method method = javaType.getRawClass().getDeclaredMethod("parseFrom", byte[].class);
        if (method != null) {
          return (T) method.invoke(null, content);
        }
      } catch (NoSuchMethodException e) {
        // It was a best effort. Skip this try.
      } catch (Exception e) {
        throw new IOException(e);
      }
    }

    // 最后才通过 jackson 进行标准的 json 序列化
    return OBJECT_MAPPER.readValue(content, javaType);
  }

private static <T> T deserializePrimitives(byte[] content, JavaType javaType) throws IOException {
    if ((content == null) || (content.length == 0)) {
      // content 为null或者空的特殊处理，相当于是缺省值
      if (javaType.hasRawClass(boolean.class)) {
        return (T) Boolean.FALSE;
      }

      if (javaType.hasRawClass(byte.class)) {
        return (T) Byte.valueOf((byte) 0);
      }

      if (javaType.hasRawClass(short.class)) {
        return (T) Short.valueOf((short) 0);
      }

      if (javaType.hasRawClass(int.class)) {
        return (T) Integer.valueOf(0);
      }

      if (javaType.hasRawClass(long.class)) {
        return (T) Long.valueOf(0L);
      }

      if (javaType.hasRawClass(float.class)) {
        return (T) Float.valueOf(0);
      }

      if (javaType.hasRawClass(double.class)) {
        return (T) Double.valueOf(0);
      }

      if (javaType.hasRawClass(char.class)) {
        return (T) Character.valueOf(Character.MIN_VALUE);
      }

      return null;
    }

    // 对于非空值，通过 jackson 进行反序列化
    return OBJECT_MAPPER.readValue(content, javaType);
  }

  public static final String API_VERSION = "v1.0";

  public static final String ALPHA_1_API_VERSION = "v1.0-alpha1";

  private static final String HEADER_DAPR_REQUEST_ID = "X-DaprRequestId";

  private static final String DEFAULT_HTTP_SCHEME = "http";

  private static final Set<String> ALLOWED_CONTEXT_IN_HEADERS =
      Collections.unmodifiableSet(new HashSet<>(Arrays.asList("grpc-trace-bin", "traceparent", "tracestate")));

  public enum HttpMethods {
    NONE,
    GET,
    PUT,
    POST,
    DELETE,
    HEAD,
    CONNECT,
    OPTIONS,
    TRACE
  }

  public static class Response {
    private byte[] body;
    private Map<String, String> headers;
    private int statusCode;
    ......
  }

  private final OkHttpClient httpClient;
  private final int port;
  private final String hostname;

  DaprHttp(String hostname, int port, OkHttpClient httpClient) {
    this.hostname = hostname;
    this.port = port;
    this.httpClient = httpClient;
  }

  /**
   * 调用API，返回文本格式有效载荷。
   *
   * @param method        HTTP method.
   * @param pathSegments  Array of path segments (/a/b/c -> ["a", "b", "c"]).
   * @param urlParameters Parameters in the URL
   * @param content       payload to be posted.
   * @param headers       HTTP headers.
   * @param context       OpenTelemetry's Context.
   * @return CompletableFuture for Response.
   */
private CompletableFuture<Response> doInvokeApi(String method,
                               String[] pathSegments,
                               Map<String, List<String>> urlParameters,
                               byte[] content, Map<String, String> headers,
                               Context context) {
    // 方法人口参数基本就是一个非常简化的HTTP请求的格式抽象

    // 取 UUID 为 requestId
    final String requestId = UUID.randomUUID().toString();
    RequestBody body;

    //组装 okhttp3 的 request
    String contentType = headers != null ? headers.get(Metadata.CONTENT_TYPE) : null;
    MediaType mediaType = contentType == null ? MEDIA_TYPE_APPLICATION_JSON : MediaType.get(contentType);
    if (content == null) {
      body = mediaType.equals(MEDIA_TYPE_APPLICATION_JSON)
          ? REQUEST_BODY_EMPTY_JSON
          : RequestBody.Companion.create(new byte[0], mediaType);
    } else {
      body = RequestBody.Companion.create(content, mediaType);
    }
    HttpUrl.Builder urlBuilder = new HttpUrl.Builder();
    urlBuilder.scheme(DEFAULT_HTTP_SCHEME)
        .host(this.hostname)
        .port(this.port);
    for (String pathSegment : pathSegments) {
      urlBuilder.addPathSegment(pathSegment);
    }
    Optional.ofNullable(urlParameters).orElse(Collections.emptyMap()).entrySet().stream()
        .forEach(urlParameter ->
            Optional.ofNullable(urlParameter.getValue()).orElse(Collections.emptyList()).stream()
              .forEach(urlParameterValue ->
                  urlBuilder.addQueryParameter(urlParameter.getKey(), urlParameterValue)));

    Request.Builder requestBuilder = new Request.Builder()
        .url(urlBuilder.build())
        .addHeader(HEADER_DAPR_REQUEST_ID, requestId);
    if (context != null) {
      context.stream()
          .filter(entry -> ALLOWED_CONTEXT_IN_HEADERS.contains(entry.getKey().toString().toLowerCase()))
          .forEach(entry -> requestBuilder.addHeader(entry.getKey().toString(), entry.getValue().toString()));
    }
    if (HttpMethods.GET.name().equals(method)) {
      requestBuilder.get();
    } else if (HttpMethods.DELETE.name().equals(method)) {
      requestBuilder.delete();
    } else {
      requestBuilder.method(method, body);
    }

    String daprApiToken = Properties.API_TOKEN.get();
    if (daprApiToken != null) {
      requestBuilder.addHeader(Headers.DAPR_API_TOKEN, daprApiToken);
    }

    if (headers != null) {
      Optional.ofNullable(headers.entrySet()).orElse(Collections.emptySet()).stream()
          .forEach(header -> {
            requestBuilder.addHeader(header.getKey(), header.getValue());
          });
    }
    // 完成 request 的组装，构建 request 对象
    Request request = requestBuilder.build();

    // 发出 okhttp3 的请求，然后返回 CompletableFuture
    CompletableFuture<Response> future = new CompletableFuture<>();
    this.httpClient.newCall(request).enqueue(new ResponseFutureCallback(future));
    return future;
  }

private DaprHttp buildDaprHttp() {
    // 双重检查锁
    if (OK_HTTP_CLIENT == null) {
      synchronized (LOCK) {
        if (OK_HTTP_CLIENT == null) {
          OkHttpClient.Builder builder = new OkHttpClient.Builder();
          Duration readTimeout = Duration.ofSeconds(Properties.HTTP_CLIENT_READ_TIMEOUT_SECONDS.get());
          builder.readTimeout(readTimeout);

          Dispatcher dispatcher = new Dispatcher();
          dispatcher.setMaxRequests(Properties.HTTP_CLIENT_MAX_REQUESTS.get());
          //这里有一个超级大坑！
          // The maximum number of requests for each host to execute concurrently.
          // Default value is 5 in okhttp which is totally UNACCEPTABLE!
          // For sidecar case, set it the same as maxRequests.
          dispatcher.setMaxRequestsPerHost(Properties.HTTP_CLIENT_MAX_REQUESTS.get());
          builder.dispatcher(dispatcher);

          ConnectionPool pool = new ConnectionPool(Properties.HTTP_CLIENT_MAX_IDLE_CONNECTIONS.get(),
                  KEEP_ALIVE_DURATION, TimeUnit.SECONDS);
          builder.connectionPool(pool);

          OK_HTTP_CLIENT = builder.build();
        }
      }
    }

    return new DaprHttp(Properties.SIDECAR_IP.get(), Properties.HTTP_PORT.get(), OK_HTTP_CLIENT);
  }
}

// 无论需要何种GRPC或HTTP客户端实现，都可以使用通用客户端适配器。
public interface DaprClient extends AutoCloseable {

  Mono<Void> waitForSidecar(int timeoutInMilliseconds);

  Mono<Void> shutdown();
}

  Mono<Void> publishEvent(String pubsubName, String topicName, Object data);

  Mono<Void> publishEvent(String pubsubName, String topicName, Object data, Map<String, String> metadata);

  Mono<Void> publishEvent(PublishEventRequest request);

// 无论需要何种GRPC或HTTP客户端实现，都可以使用通用客户端适配器。
public interface DaprPreviewClient extends AutoCloseable {

  Mono<ConfigurationItem> getConfiguration(String storeName, String key);

  Flux<List<ConfigurationItem>> subscribeToConfiguration(String storeName, String... keys);

  <T> Mono<QueryStateResponse<T>> queryState(String storeName, String query, TypeRef<T> type);
}

// 抽象类，具有客户端实现之间共同的便利方法。
abstract class AbstractDaprClient implements DaprClient, DaprPreviewClient {
  // 这里还是写死了 jackson！
  // TBD： 看下是哪里在用
  protected static final ObjectMapper JSON_REQUEST_MAPPER = new ObjectMapper();

  protected DaprObjectSerializer objectSerializer;

  protected DaprObjectSerializer stateSerializer;

    AbstractDaprClient(
      DaprObjectSerializer objectSerializer,
      DaprObjectSerializer stateSerializer) {
    this.objectSerializer = objectSerializer;
    this.stateSerializer = stateSerializer;
  }
}

  @Override
  public Mono<Void> publishEvent(String pubsubName, String topicName, Object data) {
    return this.publishEvent(pubsubName, topicName, data, null);
  }

    @Override
  public Mono<Void> publishEvent(String pubsubName, String topicName, Object data, Map<String, String> metadata) {
    PublishEventRequest req = new PublishEventRequest(pubsubName, topicName, data)
        .setMetadata(metadata);
    return this.publishEvent(req).then();
  }

public class DaprClientHttp extends AbstractDaprClient {

  private final DaprHttp client;
  private final boolean isObjectSerializerDefault;
  private final boolean isStateSerializerDefault;

  DaprClientHttp(DaprHttp client, DaprObjectSerializer objectSerializer, DaprObjectSerializer stateSerializer) {
    super(objectSerializer, stateSerializer);
    this.client = client;
    this.isObjectSerializerDefault = objectSerializer.getClass() == DefaultObjectSerializer.class;
    this.isStateSerializerDefault = stateSerializer.getClass() == DefaultObjectSerializer.class;
  }

   DaprClientHttp(DaprHttp client) {
    this(client, new DefaultObjectSerializer(), new DefaultObjectSerializer());
  }
}

  public Mono<Void> waitForSidecar(int timeoutInMilliseconds) {
    return Mono.fromRunnable(() -> {
      try {
        NetworkUtils.waitForSocket(Properties.SIDECAR_IP.get(), Properties.HTTP_PORT.get(), timeoutInMilliseconds);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    });
  }

  @Override
  public void close() {
    // 简单的关闭 http client
    client.close();
  }

@Override
  public Mono<Void> publishEvent(PublishEventRequest request) {
    try {
      String pubsubName = request.getPubsubName();
      String topic = request.getTopic();
      Object data = request.getData();
      Map<String, String> metadata = request.getMetadata();

      if (topic == null || topic.trim().isEmpty()) {
        throw new IllegalArgumentException("Topic name cannot be null or empty.");
      }

      byte[] serializedEvent = objectSerializer.serialize(data);
      // Content-type can be overwritten on a per-request basis.
      // It allows CloudEvents to be handled differently, for example.
      String contentType = request.getContentType();
      if (contentType == null || contentType.isEmpty()) {
        contentType = objectSerializer.getContentType();
      }
      Map<String, String> headers = Collections.singletonMap("content-type", contentType);

      String[] pathSegments = new String[]{ DaprHttp.API_VERSION, "publish", pubsubName, topic };

      Map<String, List<String>> queryArgs = metadataToQueryArgs(metadata);
      return Mono.subscriberContext().flatMap(
          context -> this.client.invokeApi(
              DaprHttp.HttpMethods.POST.name(), pathSegments, queryArgs, serializedEvent, headers, context
          )
      ).then();
    } catch (Exception ex) {
      return DaprException.wrapMono(ex);
    }
  }

  @Override
  public Mono<Void> shutdown() {
    String[] pathSegments = new String[]{ DaprHttp.API_VERSION, "shutdown" };
    return Mono.subscriberContext().flatMap(
            context -> client.invokeApi(DaprHttp.HttpMethods.POST.name(), pathSegments,
                null, null, context))
        .then();
  }

public class DaprClientGrpc extends AbstractDaprClient {

  private Closeable channel;

  private DaprGrpc.DaprStub asyncStub;

  DaprClientGrpc(
      Closeable closeableChannel,
      DaprGrpc.DaprStub asyncStub,
      DaprObjectSerializer objectSerializer,
      DaprObjectSerializer stateSerializer) {
    super(objectSerializer, stateSerializer);
    this.channel = closeableChannel;
    this.asyncStub = intercept(asyncStub);
  }

}

  @Override
  public Mono<Void> waitForSidecar(int timeoutInMilliseconds) {
    return Mono.fromRunnable(() -> {
      try {
        NetworkUtils.waitForSocket(Properties.SIDECAR_IP.get(), Properties.GRPC_PORT.get(), timeoutInMilliseconds);
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    });
  }

  public void close() throws Exception {
    if (channel != null) {
      DaprException.wrap(() -> {
        // 关闭channel
        channel.close();
        return true;
      }).call();
    }
  }

@Override
  public Mono<Void> publishEvent(PublishEventRequest request) {
    try {
      String pubsubName = request.getPubsubName();
      String topic = request.getTopic();
      Object data = request.getData();
      DaprProtos.PublishEventRequest.Builder envelopeBuilder = DaprProtos.PublishEventRequest.newBuilder()
          .setTopic(topic)
          .setPubsubName(pubsubName)
          .setData(ByteString.copyFrom(objectSerializer.serialize(data)));

      // Content-type can be overwritten on a per-request basis.
      // It allows CloudEvents to be handled differently, for example.
      String contentType = request.getContentType();
      if (contentType == null || contentType.isEmpty()) {
        contentType = objectSerializer.getContentType();
      }
      envelopeBuilder.setDataContentType(contentType);

      Map<String, String> metadata = request.getMetadata();
      if (metadata != null) {
        envelopeBuilder.putAllMetadata(metadata);
      }

      return Mono.subscriberContext().flatMap(
          context ->
              this.<Empty>createMono(
                  it -> intercept(context, asyncStub).publishEvent(envelopeBuilder.build(), it)
              )
      ).then();
    } catch (Exception ex) {
      return DaprException.wrapMono(ex);
    }
  }

@Documented
@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Topic {
  String name();
  String pubsubName();
  String metadata() default "{}";

  // 用于匹配传入的 cloud event 的规则。
  Rule rule() default @Rule(match = "", priority = 0);
}

  @Topic(name = "testingtopic", pubsubName = "${myAppProperty:messagebus}")
  @PostMapping(path = "/testingtopic")
  public Mono<Void> handleMessage(@RequestBody(required = false) CloudEvent<?> cloudEvent) {
    ......
  }

@Documented
@Target(ElementType.ANNOTATION_TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface Rule {

  // 用于匹配传入的 cloud event 的通用表达式语言（ Common Expression Language / CEL）表达。
  String match();

  // 规则的优先级，用于排序。最低的数字有更高的优先权。
  int priority();
}

  @Topic(name = "testingtopic", pubsubName = "${myAppProperty:messagebus}",
          rule = @Rule(match = "event.type == \"v2\"", priority = 1))
  @PostMapping(path = "/testingtopicV2")
  public Mono<Void> handleMessageV2(@RequestBody(required = false) CloudEvent cloudEvent) {
    ......
  }

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
io.dapr.springboot.DaprAutoConfiguration

@Configuration
@ConditionalOnWebApplication
@ComponentScan("io.dapr.springboot")
public class DaprAutoConfiguration {
}

@Component
public class DaprBeanPostProcessor implements BeanPostProcessor {

  private static final ObjectMapper MAPPER = new ObjectMapper();

  private final EmbeddedValueResolver embeddedValueResolver;

  DaprBeanPostProcessor(ConfigurableBeanFactory beanFactory) {
    embeddedValueResolver = new EmbeddedValueResolver(beanFactory);
  }
  ......
}

  @Override
  public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
    if (bean == null) {
      return null;
    }

    subscribeToTopics(bean.getClass(), embeddedValueResolver);

    return bean;
  }

  @Override
  public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
    return bean;
  }

@RestController
public class DaprController {
}

@GetMapping(path = "/healthz")
public void healthz() {
}

@GetMapping(path = "/dapr/config", produces = MediaType.APPLICATION_JSON_VALUE)
public byte[] daprConfig() throws IOException {
  return ActorRuntime.getInstance().serializeConfig();
}

  public byte[] serializeConfig() throws IOException {
    return INTERNAL_SERIALIZER.serialize(this.config);
  }

  private ActorRuntime(ManagedChannel channel, DaprClient daprClient) throws IllegalStateException {
    this.config = new ActorRuntimeConfig();
  }

@GetMapping(path = "/dapr/subscribe", produces = MediaType.APPLICATION_JSON_VALUE)
public byte[] daprSubscribe() throws IOException {
  return SERIALIZER.serialize(DaprRuntime.getInstance().listSubscribedTopics());
}

@DeleteMapping(path = "/actors/{type}/{id}")
  public Mono<Void> deactivateActor(@PathVariable("type") String type,
                                    @PathVariable("id") String id) {
    return ActorRuntime.getInstance().deactivate(type, id);
  }

  @PutMapping(path = "/actors/{type}/{id}/method/{method}")
  public Mono<byte[]> invokeActorMethod(@PathVariable("type") String type,
                                        @PathVariable("id") String id,
                                        @PathVariable("method") String method,
                                        @RequestBody(required = false) byte[] body) {
    return ActorRuntime.getInstance().invoke(type, id, method, body);
  }

  @PutMapping(path = "/actors/{type}/{id}/method/timer/{timer}")
  public Mono<Void> invokeActorTimer(@PathVariable("type") String type,
                                     @PathVariable("id") String id,
                                     @PathVariable("timer") String timer,
                                     @RequestBody byte[] body) {
    return ActorRuntime.getInstance().invokeTimer(type, id, timer, body);
  }

  @PutMapping(path = "/actors/{type}/{id}/method/remind/{reminder}")
  public Mono<Void> invokeActorReminder(@PathVariable("type") String type,
                                        @PathVariable("id") String id,
                                        @PathVariable("reminder") String reminder,
                                        @RequestBody(required = false) byte[] body) {
    return ActorRuntime.getInstance().invokeReminder(type, id, reminder, body);
  }

  @Topic(name = "testingtopic", pubsubName = "${myAppProperty:messagebus}",
          rule = @Rule(match = "event.type == \"v2\"", priority = 1))
  @PostMapping(path = "/testingtopicV2")
  public Mono<Void> handleMessageV2(@RequestBody(required = false) CloudEvent cloudEvent) {
    ......
  }

@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
  subscribeToTopics(bean.getClass(), embeddedValueResolver);
  return bean;
}

private static void subscribeToTopics(Class clazz, EmbeddedValueResolver embeddedValueResolver) {
    if (clazz == null) {
      return;
    }

    // 先用 Superclass 做一次递归调用，这样就会从当前类的父类开始先推衍
    // 由于每次都是父类先执行，因此这会一直递归到最顶层的 Object 类 
    subscribeToTopics(clazz.getSuperclass(), embeddedValueResolver);
    // 取当前类的所有方法
    for (Method method : clazz.getDeclaredMethods()) {
      // 然后看方法上是不是标记了 dapr 的 topic 注解
      Topic topic = method.getAnnotation(Topic.class);
      if (topic == null) {
        continue;
      }

      // 如果方法上有标记 dapr 的 topic 注解，则开始处理
      // 先获取 topic 注解上的属性 topic name, pubsub name, rule 
      Rule rule = topic.rule();
      String topicName = embeddedValueResolver.resolveStringValue(topic.name());
      String pubSubName = embeddedValueResolver.resolveStringValue(topic.pubsubName());
      // rule 也是一个注解，获取 match 属性
      String match = embeddedValueResolver.resolveStringValue(rule.match());
      if ((topicName != null) && (topicName.length() > 0) && pubSubName != null && pubSubName.length() > 0) {
        // topicName 和 pubSubName 不能为空 （metadata 可以为空，rule可以为空）
        try {
          TypeReference<HashMap<String, String>> typeRef
                  = new TypeReference<HashMap<String, String>>() {};
          // 读取 topic 注解上的 metadata 属性
          Map<String, String> metadata = MAPPER.readValue(topic.metadata(), typeRef);
          // 读取路由信息，细节看下一节
          List<String> routes = getAllCompleteRoutesForPost(clazz, method, topicName);
          for (String route : routes) {
            // 将读取的路由信息添加到 dapr runtime 中。
            // 细节看下一节
            DaprRuntime.getInstance().addSubscribedTopic(
                pubSubName, topicName, match, rule.priority(), route, metadata);
          }
        } catch (JsonProcessingException e) {
          throw new IllegalArgumentException("Error while parsing metadata: " + e);
        }
      }
    }
  }