在事件总线上交换生成的 Protobuf 类

syntax = "proto3";

package greeting;

option java_multiple_files = true;
option java_package = "io.vertx.howtos.protobuf.eventbus";
option java_outer_classname = "GreetingProtos";

message GreetingRequest {
  string name = 1;
}

message GreetingReply {
  string message = 1;
}

package io.vertx.howtos.protobuf.eventbus;

import io.vertx.core.Future;
import io.vertx.core.VerticleBase;

public class ReceiverVerticle extends VerticleBase {

  @Override
  public Future<?> start() {
    return vertx.eventBus().<GreetingRequest>consumer("greetings", msg -> {
      var request = msg.body();
      System.out.printf("Received request = %s (%d)%n", request.getName(), System.identityHashCode(request));
      var greeting = String.format("Hello %s", request.getName());
      var reply = GreetingReply.newBuilder().setMessage(greeting).build();
      System.out.printf("Sending reply = %s (%d)%n", reply.getMessage(), System.identityHashCode(reply));
      msg.reply(reply);
    }).completion();
  }
}

package io.vertx.howtos.protobuf.eventbus;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.eventbus.Message;

public class SenderVerticle extends AbstractVerticle {

  @Override
  public void start() throws Exception {
    vertx.setPeriodic(5000, l -> {
      var request = GreetingRequest.newBuilder().setName("Jane Doe").build();
      System.out.printf("Sending request = %s (%d)%n", request.getName(), System.identityHashCode(request));
      vertx.eventBus().<GreetingReply>request("greetings", request)
        .map(Message::body)
        .onFailure(Throwable::printStackTrace)
        .onSuccess(reply -> System.out.printf("Received reply = %s (%d)%n", reply.getMessage(), System.identityHashCode(reply)));
    });
  }
}

package io.vertx.howtos.protobuf.eventbus;

import com.google.protobuf.GeneratedMessageV3;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.MessageCodec;
import io.vertx.core.impl.SerializableUtils;

public class ProtobufCodec implements MessageCodec<GeneratedMessageV3, GeneratedMessageV3> {

  static final String PROTOS_PACKAGE_NAME = "io.vertx.howtos.protobuf.eventbus.";

  @Override
  public void encodeToWire(Buffer buffer, GeneratedMessageV3 o) {
    var bytes = SerializableUtils.toBytes(o);
    buffer.appendInt(bytes.length);
    buffer.appendBytes(bytes);
  }

  @Override
  public GeneratedMessageV3 decodeFromWire(int pos, Buffer buffer) {
    var length = buffer.getInt(pos);
    pos += 4;
    var bytes = buffer.getBytes(pos, pos + length);
    return (GeneratedMessageV3) SerializableUtils.fromBytes(bytes, CheckedClassNameObjectInputStream::new);
  }

  @Override
  public GeneratedMessageV3 transform(GeneratedMessageV3 o) {
    return o;
  }

  @Override
  public String name() {
    return "ProtobufCodec";
  }

  @Override
  public byte systemCodecID() {
    return -1; // -1 for a user codec
  }

  public boolean appliesTo(String className) {
    return className.startsWith(PROTOS_PACKAGE_NAME);
  }
}

package io.vertx.howtos.protobuf.eventbus;

import io.vertx.core.eventbus.EventBus;

import java.io.*;

class CheckedClassNameObjectInputStream extends ObjectInputStream {

  CheckedClassNameObjectInputStream(InputStream in) throws IOException {
    super(in);
  }

  @Override
  protected Class<?> resolveClass(ObjectStreamClass desc) throws IOException, ClassNotFoundException {
    var name = desc.getName();
    if (name.startsWith("com.google.protobuf.")
      || name.startsWith(ProtobufCodec.PROTOS_PACKAGE_NAME)
      || EventBus.DEFAULT_SERIALIZABLE_CHECKER.apply(name)) {
      return super.resolveClass(desc);
    }
    throw new InvalidClassException("Class not allowed: " + name);
  }
}

package io.vertx.howtos.protobuf.eventbus;

import io.vertx.launcher.application.HookContext;
import io.vertx.launcher.application.VertxApplication;
import io.vertx.launcher.application.VertxApplicationHooks;

public class CustomLauncher extends VertxApplication implements VertxApplicationHooks {

  public CustomLauncher(String[] args) {
    super(args);
  }

  public static void main(String[] args) {
    new CustomLauncher(args).launch();
  }

  @Override
  public void afterVertxStarted(HookContext context) {
    var vertx = context.vertx();
    var protobufCodec = new ProtobufCodec();
    vertx.eventBus().registerCodec(protobufCodec);
    vertx.eventBus().codecSelector(body -> {
      return protobufCodec.appliesTo(body.getClass().getName()) ? protobufCodec.name() : null;
    });
  }
}