class DaprStateAsyncProvider {

  Mono<Void> apply(String actorType, ActorId actorId, ActorStateChange... stateChanges) {
    if ((stateChanges == null) || stateChanges.length == 0) {
      return Mono.empty();
    }

    int count = 0;
    // Constructing the JSON via a stream API to avoid creating transient objects to be instantiated.
    byte[] payload = null;
    try (ByteArrayOutputStream writer = new ByteArrayOutputStream()) {
      JsonGenerator generator = JSON_FACTORY.createGenerator(writer);
      // Start array
      generator.writeStartArray();

      for (ActorStateChange stateChange : stateChanges) {
        if ((stateChange == null) || (stateChange.getChangeKind() == null)) {
          continue;
        }

        String operationName = stateChange.getChangeKind().getDaprStateChangeOperation();
        if ((operationName == null) || (operationName.length() == 0)) {
          continue;
        }

        count++;

        // Start operation object.
        generator.writeStartObject();
        generator.writeStringField("operation", operationName);

        // Start request object.
        generator.writeObjectFieldStart("request");
        generator.writeStringField("key", stateChange.getStateName());
        if ((stateChange.getChangeKind() == ActorStateChangeKind.UPDATE)
            || (stateChange.getChangeKind() == ActorStateChangeKind.ADD)) {
          byte[] data = this.stateSerializer.serialize(stateChange.getValue());
          if (data != null) {
            if (this.isStateSerializerDefault) {
              // DefaultObjectSerializer is a JSON serializer, so we just pass it on.
              generator.writeFieldName("value");
              generator.writeRawValue(new String(data, CHARSET));
            } else {
              // Custom serializer uses byte[].
              generator.writeBinaryField("value", data);
            }
          }
        }
        // End request object.
        generator.writeEndObject();

        // End operation object.
        generator.writeEndObject();
      }

      // End array
      generator.writeEndArray();

      generator.close();
      writer.flush();
      payload = writer.toByteArray();
    } catch (IOException e) {
      e.printStackTrace();
      return Mono.error(e);
    }

    if (count == 0) {
      // No-op since there is no operation to be performed.
      Mono.empty();
    }

    return this.daprClient.saveActorStateTransactionally(actorType, actorId.toString(), payload);
  }

    DaprStateAsyncProvider(DaprClient daprClient, DaprObjectSerializer stateSerializer);

}