diff --git a/spring-pulsar-dependencies/build.gradle b/spring-pulsar-dependencies/build.gradle index 698ccd5c..bd765e38 100644 --- a/spring-pulsar-dependencies/build.gradle +++ b/spring-pulsar-dependencies/build.gradle @@ -9,21 +9,22 @@ javaPlatform { ext { assertjVersion = '3.23.1' awaitilityVersion = '4.2.0' + caffeineVersion = '3.1.1' googleJsr305Version = '3.0.2' hamcrestVersion = '2.2' hibernateValidationVersion = '7.0.4.Final' jacksonBomVersion = '2.13.3' jaywayJsonPathVersion = '2.6.0' junitJupiterVersion = '5.9.0' - pulsarVersion = '2.10.1' log4jVersion = '2.18.0' mockitoVersion = '4.6.1' - reactorVersion = '2020.0.17' + protobufJavaVersion = '3.21.5' pulsarTestcontainersVersion = '1.17.3' + pulsarVersion = '2.10.1' + reactorVersion = '2020.0.17' springBootVersion = '3.0.0-SNAPSHOT' springRetryVersion = '1.3.3' springVersion = '6.0.0-SNAPSHOT' - caffeineVersion = '3.1.1' } dependencies { @@ -38,6 +39,7 @@ dependencies { constraints { api "com.github.ben-manes.caffeine:caffeine:$caffeineVersion" api "com.google.code.findbugs:jsr305:$googleJsr305Version" + api "com.google.protobuf:protobuf-java:$protobufJavaVersion" api "com.jayway.jsonpath:json-path:$jaywayJsonPathVersion" api "org.apache.pulsar:pulsar-client:$pulsarVersion" api "org.apache.pulsar:pulsar-client-admin:$pulsarVersion" diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc index 5d7541ff..5dbfb529 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc @@ -639,8 +639,16 @@ public void listen(Foo foo) { ---- ==== -On the producer side also, for the Java primitive types, the framework can infer the Schema, but for any other types, you need set that on the `PulsarTemmplate`. +On the producer side also, for the Java primitive types, the framework can infer the Schema, but for any other types, you need to set them on the `PulsarTemplate` as shown below. +==== +[source, java] +---- +template.setSchema(JSONSchema.of(Foo.class)); +---- +==== + +TIP: Complex Schema types that are currently supported are JSON, AVRO, PROTOBUF, and KEY_VALUE. For KEY_VALUE schemata, only INLINE encoding is supported. ==== Intercepting messages diff --git a/spring-pulsar/build.gradle b/spring-pulsar/build.gradle index e413c202..6175179e 100644 --- a/spring-pulsar/build.gradle +++ b/spring-pulsar/build.gradle @@ -6,6 +6,7 @@ description = 'Spring Pulsar Support' dependencies { api 'com.github.ben-manes.caffeine:caffeine' + api 'com.google.protobuf:protobuf-java' api 'org.apache.pulsar:pulsar-client' api 'org.apache.pulsar:pulsar-client-admin' api 'org.apache.pulsar:pulsar-client-admin-api' diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java index 63ca9311..9b012d06 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/config/MethodPulsarListenerEndpoint.java @@ -30,6 +30,7 @@ import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; +import org.apache.pulsar.client.impl.schema.ProtobufSchema; import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaType; @@ -55,6 +56,8 @@ import org.springframework.pulsar.support.converter.PulsarBatchMessageConverter; import org.springframework.pulsar.support.converter.PulsarRecordMessageConverter; import org.springframework.util.Assert; +import com.google.protobuf.GeneratedMessageV3; + /** * A {@link PulsarListenerEndpoint} providing the method to invoke to process an incoming * message for this endpoint. @@ -157,6 +160,12 @@ public class MethodPulsarListenerEndpoint extends AbstractPulsarListenerEndpo Schema messageSchema = getMessageSchema(messageParameter, AvroSchema::of); pulsarContainerProperties.setSchema(messageSchema); } + case PROTOBUF -> { + @SuppressWarnings("unchecked") + Schema messageSchema = getMessageSchema(messageParameter, + (c -> ProtobufSchema.of((Class) c))); + pulsarContainerProperties.setSchema(messageSchema); + } case KEY_VALUE -> { Schema messageSchema = getMessageKeyValueSchema(messageParameter); pulsarContainerProperties.setSchema(messageSchema); diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/Proto.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/Proto.java new file mode 100644 index 00000000..40cabbba --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/Proto.java @@ -0,0 +1,765 @@ +// Generated by the protocol buffer compiler. DO NOT EDIT! +// source: person.proto + +package org.springframework.pulsar.listener; + +public final class Proto { + + private Proto() { + } + + public static void registerAllExtensions(com.google.protobuf.ExtensionRegistryLite registry) { + } + + public static void registerAllExtensions(com.google.protobuf.ExtensionRegistry registry) { + registerAllExtensions((com.google.protobuf.ExtensionRegistryLite) registry); + } + + public interface PersonOrBuilder extends + // @@protoc_insertion_point(interface_extends:proto.Person) + com.google.protobuf.MessageOrBuilder { + + /** + * optional int32 id = 1; + * @return Whether the id field is set. + */ + boolean hasId(); + + /** + * optional int32 id = 1; + * @return The id. + */ + int getId(); + + /** + * optional string name = 2; + * @return Whether the name field is set. + */ + boolean hasName(); + + /** + * optional string name = 2; + * @return The name. + */ + java.lang.String getName(); + + /** + * optional string name = 2; + * @return The bytes for name. + */ + com.google.protobuf.ByteString getNameBytes(); + + } + + /** + * Protobuf type {@code proto.Person} + */ + public static final class Person extends com.google.protobuf.GeneratedMessageV3 implements + // @@protoc_insertion_point(message_implements:proto.Person) + PersonOrBuilder { + + private static final long serialVersionUID = 0L; + + // Use Person.newBuilder() to construct. + private Person(com.google.protobuf.GeneratedMessageV3.Builder builder) { + super(builder); + } + + private Person() { + name_ = ""; + } + + @java.lang.Override + @SuppressWarnings({ "unused" }) + protected java.lang.Object newInstance(UnusedPrivateParameter unused) { + return new Person(); + } + + @java.lang.Override + public final com.google.protobuf.UnknownFieldSet getUnknownFields() { + return this.unknownFields; + } + + private Person(com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + this(); + if (extensionRegistry == null) { + throw new java.lang.NullPointerException(); + } + int mutable_bitField0_ = 0; + com.google.protobuf.UnknownFieldSet.Builder unknownFields = com.google.protobuf.UnknownFieldSet + .newBuilder(); + try { + boolean done = false; + while (!done) { + int tag = input.readTag(); + switch (tag) { + case 0: + done = true; + break; + case 8: { + bitField0_ |= 0x00000001; + id_ = input.readInt32(); + break; + } + case 18: { + java.lang.String s = input.readStringRequireUtf8(); + bitField0_ |= 0x00000002; + name_ = s; + break; + } + default: { + if (!parseUnknownField(input, unknownFields, extensionRegistry, tag)) { + done = true; + } + break; + } + } + } + } + catch (com.google.protobuf.InvalidProtocolBufferException e) { + throw e.setUnfinishedMessage(this); + } + catch (com.google.protobuf.UninitializedMessageException e) { + throw e.asInvalidProtocolBufferException().setUnfinishedMessage(this); + } + catch (java.io.IOException e) { + throw new com.google.protobuf.InvalidProtocolBufferException(e).setUnfinishedMessage(this); + } + finally { + this.unknownFields = unknownFields.build(); + makeExtensionsImmutable(); + } + } + + public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() { + return org.springframework.pulsar.listener.Proto.internal_static_proto_Person_descriptor; + } + + @java.lang.Override + protected com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internalGetFieldAccessorTable() { + return org.springframework.pulsar.listener.Proto.internal_static_proto_Person_fieldAccessorTable + .ensureFieldAccessorsInitialized(org.springframework.pulsar.listener.Proto.Person.class, + org.springframework.pulsar.listener.Proto.Person.Builder.class); + } + + private int bitField0_; + + public static final int ID_FIELD_NUMBER = 1; + + private int id_; + + /** + * optional int32 id = 1; + * @return Whether the id field is set. + */ + @java.lang.Override + public boolean hasId() { + return ((bitField0_ & 0x00000001) != 0); + } + + /** + * optional int32 id = 1; + * @return The id. + */ + @java.lang.Override + public int getId() { + return id_; + } + + public static final int NAME_FIELD_NUMBER = 2; + + private volatile java.lang.Object name_; + + /** + * optional string name = 2; + * @return Whether the name field is set. + */ + @java.lang.Override + public boolean hasName() { + return ((bitField0_ & 0x00000002) != 0); + } + + /** + * optional string name = 2; + * @return The name. + */ + @java.lang.Override + public java.lang.String getName() { + java.lang.Object ref = name_; + if (ref instanceof java.lang.String) { + return (java.lang.String) ref; + } + else { + com.google.protobuf.ByteString bs = (com.google.protobuf.ByteString) ref; + java.lang.String s = bs.toStringUtf8(); + name_ = s; + return s; + } + } + + /** + * optional string name = 2; + * @return The bytes for name. + */ + @java.lang.Override + public com.google.protobuf.ByteString getNameBytes() { + java.lang.Object ref = name_; + if (ref instanceof java.lang.String) { + com.google.protobuf.ByteString b = com.google.protobuf.ByteString.copyFromUtf8((java.lang.String) ref); + name_ = b; + return b; + } + else { + return (com.google.protobuf.ByteString) ref; + } + } + + private byte memoizedIsInitialized = -1; + + @java.lang.Override + public final boolean isInitialized() { + byte isInitialized = memoizedIsInitialized; + if (isInitialized == 1) + return true; + if (isInitialized == 0) + return false; + + memoizedIsInitialized = 1; + return true; + } + + @java.lang.Override + public void writeTo(com.google.protobuf.CodedOutputStream output) throws java.io.IOException { + if (((bitField0_ & 0x00000001) != 0)) { + output.writeInt32(1, id_); + } + if (((bitField0_ & 0x00000002) != 0)) { + com.google.protobuf.GeneratedMessageV3.writeString(output, 2, name_); + } + unknownFields.writeTo(output); + } + + @java.lang.Override + public int getSerializedSize() { + int size = memoizedSize; + if (size != -1) + return size; + + size = 0; + if (((bitField0_ & 0x00000001) != 0)) { + size += com.google.protobuf.CodedOutputStream.computeInt32Size(1, id_); + } + if (((bitField0_ & 0x00000002) != 0)) { + size += com.google.protobuf.GeneratedMessageV3.computeStringSize(2, name_); + } + size += unknownFields.getSerializedSize(); + memoizedSize = size; + return size; + } + + @java.lang.Override + public boolean equals(final java.lang.Object obj) { + if (obj == this) { + return true; + } + if (!(obj instanceof org.springframework.pulsar.listener.Proto.Person)) { + return super.equals(obj); + } + org.springframework.pulsar.listener.Proto.Person other = (org.springframework.pulsar.listener.Proto.Person) obj; + + if (hasId() != other.hasId()) + return false; + if (hasId()) { + if (getId() != other.getId()) + return false; + } + if (hasName() != other.hasName()) + return false; + if (hasName()) { + if (!getName().equals(other.getName())) + return false; + } + if (!unknownFields.equals(other.unknownFields)) + return false; + return true; + } + + @java.lang.Override + public int hashCode() { + if (memoizedHashCode != 0) { + return memoizedHashCode; + } + int hash = 41; + hash = (19 * hash) + getDescriptor().hashCode(); + if (hasId()) { + hash = (37 * hash) + ID_FIELD_NUMBER; + hash = (53 * hash) + getId(); + } + if (hasName()) { + hash = (37 * hash) + NAME_FIELD_NUMBER; + hash = (53 * hash) + getName().hashCode(); + } + hash = (29 * hash) + unknownFields.hashCode(); + memoizedHashCode = hash; + return hash; + } + + public static org.springframework.pulsar.listener.Proto.Person parseFrom(java.nio.ByteBuffer data) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data); + } + + public static org.springframework.pulsar.listener.Proto.Person parseFrom(java.nio.ByteBuffer data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data, extensionRegistry); + } + + public static org.springframework.pulsar.listener.Proto.Person parseFrom(com.google.protobuf.ByteString data) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data); + } + + public static org.springframework.pulsar.listener.Proto.Person parseFrom(com.google.protobuf.ByteString data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data, extensionRegistry); + } + + public static org.springframework.pulsar.listener.Proto.Person parseFrom(byte[] data) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data); + } + + public static org.springframework.pulsar.listener.Proto.Person parseFrom(byte[] data, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return PARSER.parseFrom(data, extensionRegistry); + } + + public static org.springframework.pulsar.listener.Proto.Person parseFrom(java.io.InputStream input) + throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input); + } + + public static org.springframework.pulsar.listener.Proto.Person parseFrom(java.io.InputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input, extensionRegistry); + } + + public static org.springframework.pulsar.listener.Proto.Person parseDelimitedFrom(java.io.InputStream input) + throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseDelimitedWithIOException(PARSER, input); + } + + public static org.springframework.pulsar.listener.Proto.Person parseDelimitedFrom(java.io.InputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseDelimitedWithIOException(PARSER, input, + extensionRegistry); + } + + public static org.springframework.pulsar.listener.Proto.Person parseFrom( + com.google.protobuf.CodedInputStream input) throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input); + } + + public static org.springframework.pulsar.listener.Proto.Person parseFrom( + com.google.protobuf.CodedInputStream input, com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws java.io.IOException { + return com.google.protobuf.GeneratedMessageV3.parseWithIOException(PARSER, input, extensionRegistry); + } + + @java.lang.Override + public Builder newBuilderForType() { + return newBuilder(); + } + + public static Builder newBuilder() { + return DEFAULT_INSTANCE.toBuilder(); + } + + public static Builder newBuilder(org.springframework.pulsar.listener.Proto.Person prototype) { + return DEFAULT_INSTANCE.toBuilder().mergeFrom(prototype); + } + + @java.lang.Override + public Builder toBuilder() { + return this == DEFAULT_INSTANCE ? new Builder() : new Builder().mergeFrom(this); + } + + @java.lang.Override + protected Builder newBuilderForType(com.google.protobuf.GeneratedMessageV3.BuilderParent parent) { + Builder builder = new Builder(parent); + return builder; + } + + /** + * Protobuf type {@code proto.Person} + */ + public static final class Builder extends com.google.protobuf.GeneratedMessageV3.Builder implements + // @@protoc_insertion_point(builder_implements:proto.Person) + org.springframework.pulsar.listener.Proto.PersonOrBuilder { + + public static final com.google.protobuf.Descriptors.Descriptor getDescriptor() { + return org.springframework.pulsar.listener.Proto.internal_static_proto_Person_descriptor; + } + + @java.lang.Override + protected com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internalGetFieldAccessorTable() { + return org.springframework.pulsar.listener.Proto.internal_static_proto_Person_fieldAccessorTable + .ensureFieldAccessorsInitialized(org.springframework.pulsar.listener.Proto.Person.class, + org.springframework.pulsar.listener.Proto.Person.Builder.class); + } + + // Construct using + // org.springframework.pulsar.listener.Proto.Person.newBuilder() + private Builder() { + maybeForceBuilderInitialization(); + } + + private Builder(com.google.protobuf.GeneratedMessageV3.BuilderParent parent) { + super(parent); + maybeForceBuilderInitialization(); + } + + private void maybeForceBuilderInitialization() { + if (com.google.protobuf.GeneratedMessageV3.alwaysUseFieldBuilders) { + } + } + + @java.lang.Override + public Builder clear() { + super.clear(); + id_ = 0; + bitField0_ = (bitField0_ & ~0x00000001); + name_ = ""; + bitField0_ = (bitField0_ & ~0x00000002); + return this; + } + + @java.lang.Override + public com.google.protobuf.Descriptors.Descriptor getDescriptorForType() { + return org.springframework.pulsar.listener.Proto.internal_static_proto_Person_descriptor; + } + + @java.lang.Override + public org.springframework.pulsar.listener.Proto.Person getDefaultInstanceForType() { + return org.springframework.pulsar.listener.Proto.Person.getDefaultInstance(); + } + + @java.lang.Override + public org.springframework.pulsar.listener.Proto.Person build() { + org.springframework.pulsar.listener.Proto.Person result = buildPartial(); + if (!result.isInitialized()) { + throw newUninitializedMessageException(result); + } + return result; + } + + @java.lang.Override + public org.springframework.pulsar.listener.Proto.Person buildPartial() { + org.springframework.pulsar.listener.Proto.Person result = new org.springframework.pulsar.listener.Proto.Person( + this); + int from_bitField0_ = bitField0_; + int to_bitField0_ = 0; + if (((from_bitField0_ & 0x00000001) != 0)) { + result.id_ = id_; + to_bitField0_ |= 0x00000001; + } + if (((from_bitField0_ & 0x00000002) != 0)) { + to_bitField0_ |= 0x00000002; + } + result.name_ = name_; + result.bitField0_ = to_bitField0_; + onBuilt(); + return result; + } + + @java.lang.Override + public Builder clone() { + return super.clone(); + } + + @java.lang.Override + public Builder setField(com.google.protobuf.Descriptors.FieldDescriptor field, java.lang.Object value) { + return super.setField(field, value); + } + + @java.lang.Override + public Builder clearField(com.google.protobuf.Descriptors.FieldDescriptor field) { + return super.clearField(field); + } + + @java.lang.Override + public Builder clearOneof(com.google.protobuf.Descriptors.OneofDescriptor oneof) { + return super.clearOneof(oneof); + } + + @java.lang.Override + public Builder setRepeatedField(com.google.protobuf.Descriptors.FieldDescriptor field, int index, + java.lang.Object value) { + return super.setRepeatedField(field, index, value); + } + + @java.lang.Override + public Builder addRepeatedField(com.google.protobuf.Descriptors.FieldDescriptor field, + java.lang.Object value) { + return super.addRepeatedField(field, value); + } + + @java.lang.Override + public Builder mergeFrom(com.google.protobuf.Message other) { + if (other instanceof org.springframework.pulsar.listener.Proto.Person) { + return mergeFrom((org.springframework.pulsar.listener.Proto.Person) other); + } + else { + super.mergeFrom(other); + return this; + } + } + + public Builder mergeFrom(org.springframework.pulsar.listener.Proto.Person other) { + if (other == org.springframework.pulsar.listener.Proto.Person.getDefaultInstance()) + return this; + if (other.hasId()) { + setId(other.getId()); + } + if (other.hasName()) { + bitField0_ |= 0x00000002; + name_ = other.name_; + onChanged(); + } + this.mergeUnknownFields(other.unknownFields); + onChanged(); + return this; + } + + @java.lang.Override + public final boolean isInitialized() { + return true; + } + + @java.lang.Override + public Builder mergeFrom(com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) throws java.io.IOException { + org.springframework.pulsar.listener.Proto.Person parsedMessage = null; + try { + parsedMessage = PARSER.parsePartialFrom(input, extensionRegistry); + } + catch (com.google.protobuf.InvalidProtocolBufferException e) { + parsedMessage = (org.springframework.pulsar.listener.Proto.Person) e.getUnfinishedMessage(); + throw e.unwrapIOException(); + } + finally { + if (parsedMessage != null) { + mergeFrom(parsedMessage); + } + } + return this; + } + + private int bitField0_; + + private int id_; + + /** + * optional int32 id = 1; + * @return Whether the id field is set. + */ + @java.lang.Override + public boolean hasId() { + return ((bitField0_ & 0x00000001) != 0); + } + + /** + * optional int32 id = 1; + * @return The id. + */ + @java.lang.Override + public int getId() { + return id_; + } + + /** + * optional int32 id = 1; + * @param value The id to set. + * @return This builder for chaining. + */ + public Builder setId(int value) { + bitField0_ |= 0x00000001; + id_ = value; + onChanged(); + return this; + } + + /** + * optional int32 id = 1; + * @return This builder for chaining. + */ + public Builder clearId() { + bitField0_ = (bitField0_ & ~0x00000001); + id_ = 0; + onChanged(); + return this; + } + + private java.lang.Object name_ = ""; + + /** + * optional string name = 2; + * @return Whether the name field is set. + */ + public boolean hasName() { + return ((bitField0_ & 0x00000002) != 0); + } + + /** + * optional string name = 2; + * @return The name. + */ + public java.lang.String getName() { + java.lang.Object ref = name_; + if (!(ref instanceof java.lang.String)) { + com.google.protobuf.ByteString bs = (com.google.protobuf.ByteString) ref; + java.lang.String s = bs.toStringUtf8(); + name_ = s; + return s; + } + else { + return (java.lang.String) ref; + } + } + + /** + * optional string name = 2; + * @return The bytes for name. + */ + public com.google.protobuf.ByteString getNameBytes() { + java.lang.Object ref = name_; + if (ref instanceof String) { + com.google.protobuf.ByteString b = com.google.protobuf.ByteString + .copyFromUtf8((java.lang.String) ref); + name_ = b; + return b; + } + else { + return (com.google.protobuf.ByteString) ref; + } + } + + /** + * optional string name = 2; + * @param value The name to set. + * @return This builder for chaining. + */ + public Builder setName(java.lang.String value) { + if (value == null) { + throw new NullPointerException(); + } + bitField0_ |= 0x00000002; + name_ = value; + onChanged(); + return this; + } + + /** + * optional string name = 2; + * @return This builder for chaining. + */ + public Builder clearName() { + bitField0_ = (bitField0_ & ~0x00000002); + name_ = getDefaultInstance().getName(); + onChanged(); + return this; + } + + /** + * optional string name = 2; + * @param value The bytes for name to set. + * @return This builder for chaining. + */ + public Builder setNameBytes(com.google.protobuf.ByteString value) { + if (value == null) { + throw new NullPointerException(); + } + checkByteStringIsUtf8(value); + bitField0_ |= 0x00000002; + name_ = value; + onChanged(); + return this; + } + + @java.lang.Override + public final Builder setUnknownFields(final com.google.protobuf.UnknownFieldSet unknownFields) { + return super.setUnknownFields(unknownFields); + } + + @java.lang.Override + public final Builder mergeUnknownFields(final com.google.protobuf.UnknownFieldSet unknownFields) { + return super.mergeUnknownFields(unknownFields); + } + + // @@protoc_insertion_point(builder_scope:proto.Person) + + } + + // @@protoc_insertion_point(class_scope:proto.Person) + private static final org.springframework.pulsar.listener.Proto.Person DEFAULT_INSTANCE; + static { + DEFAULT_INSTANCE = new org.springframework.pulsar.listener.Proto.Person(); + } + + public static org.springframework.pulsar.listener.Proto.Person getDefaultInstance() { + return DEFAULT_INSTANCE; + } + + private static final com.google.protobuf.Parser PARSER = new com.google.protobuf.AbstractParser() { + @java.lang.Override + public Person parsePartialFrom(com.google.protobuf.CodedInputStream input, + com.google.protobuf.ExtensionRegistryLite extensionRegistry) + throws com.google.protobuf.InvalidProtocolBufferException { + return new Person(input, extensionRegistry); + } + }; + + public static com.google.protobuf.Parser parser() { + return PARSER; + } + + @java.lang.Override + public com.google.protobuf.Parser getParserForType() { + return PARSER; + } + + @java.lang.Override + public org.springframework.pulsar.listener.Proto.Person getDefaultInstanceForType() { + return DEFAULT_INSTANCE; + } + + } + + private static final com.google.protobuf.Descriptors.Descriptor internal_static_proto_Person_descriptor; + + private static final com.google.protobuf.GeneratedMessageV3.FieldAccessorTable internal_static_proto_Person_fieldAccessorTable; + + public static com.google.protobuf.Descriptors.FileDescriptor getDescriptor() { + return descriptor; + } + + private static com.google.protobuf.Descriptors.FileDescriptor descriptor; + static { + java.lang.String[] descriptorData = { "\n\014person.proto\022\005proto\"<\n\006Person\022\017\n\002id\030\001 " + + "\001(\005H\000\210\001\001\022\021\n\004name\030\002 \001(\tH\001\210\001\001B\005\n\003_idB\007\n\005_n" + + "ameB,\n#org.springframework.pulsar.listen" + "erB\005Protob\006proto3" }; + descriptor = com.google.protobuf.Descriptors.FileDescriptor.internalBuildGeneratedFileFrom(descriptorData, + new com.google.protobuf.Descriptors.FileDescriptor[] {}); + internal_static_proto_Person_descriptor = getDescriptor().getMessageTypes().get(0); + internal_static_proto_Person_fieldAccessorTable = new com.google.protobuf.GeneratedMessageV3.FieldAccessorTable( + internal_static_proto_Person_descriptor, new java.lang.String[] { "Id", "Name", "Id", "Name", }); + } + + // @@protoc_insertion_point(outer_class_scope) + +} diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java index b7ea37a0..76ce6fea 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/listener/PulsarListenerTests.java @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Properties; @@ -28,12 +29,14 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.api.Messages; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.RedeliveryBackoff; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff; import org.apache.pulsar.client.impl.schema.AvroSchema; import org.apache.pulsar.client.impl.schema.JSONSchema; +import org.apache.pulsar.client.impl.schema.ProtobufSchema; import org.apache.pulsar.common.schema.KeyValue; import org.apache.pulsar.common.schema.KeyValueEncodingType; import org.apache.pulsar.common.schema.SchemaType; @@ -290,11 +293,14 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { @ContextConfiguration(classes = SchemaTestCases.SchemaTestConfig.class) class SchemaTestCases { - static CountDownLatch jsonLatch = new CountDownLatch(1); - - static CountDownLatch avroLatch = new CountDownLatch(1); - - static CountDownLatch keyvalueLatch = new CountDownLatch(1); + static CountDownLatch jsonLatch = new CountDownLatch(3); + static CountDownLatch jsonBatchLatch = new CountDownLatch(3); + static CountDownLatch avroLatch = new CountDownLatch(3); + static CountDownLatch avroBatchLatch = new CountDownLatch(3); + static CountDownLatch keyvalueLatch = new CountDownLatch(3); + static CountDownLatch keyvalueBatchLatch = new CountDownLatch(3); + static CountDownLatch protobufLatch = new CountDownLatch(3); + static CountDownLatch protobufBatchLatch = new CountDownLatch(3); @Test void jsonSchema() throws Exception { @@ -302,8 +308,12 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { Collections.emptyMap()); PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); template.setSchema(JSONSchema.of(User.class)); - template.send("json-topic", new User("Jason", 1)); + + for (int i = 0; i < 3; i++) { + template.send("json-topic", new User("Jason", i)); + } assertThat(jsonLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(jsonBatchLatch.await(10, TimeUnit.SECONDS)).isTrue(); } @Test @@ -312,8 +322,12 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { Collections.emptyMap()); PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); template.setSchema(AvroSchema.of(User.class)); - template.send("avro-topic", new User("Avi", 2)); + + for (int i = 0; i < 3; i++) { + template.send("avro-topic", new User("Avi", i)); + } assertThat(avroLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(avroBatchLatch.await(10, TimeUnit.SECONDS)).isTrue(); } @Test @@ -321,13 +335,29 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { PulsarProducerFactory> pulsarProducerFactory = new DefaultPulsarProducerFactory<>( pulsarClient, Collections.emptyMap()); PulsarTemplate> template = new PulsarTemplate<>(pulsarProducerFactory); - Schema> kvSchema = Schema.KeyValue(Schema.STRING, Schema.INT32, KeyValueEncodingType.INLINE); - template.setSchema(kvSchema); - template.send("keyvalue-topic", new KeyValue<>("Kevin", 3)); + + for (int i = 0; i < 3; i++) { + template.send("keyvalue-topic", new KeyValue<>("Kevin", i)); + } assertThat(keyvalueLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(keyvalueBatchLatch.await(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void protobufSchema() throws Exception { + PulsarProducerFactory pulsarProducerFactory = new DefaultPulsarProducerFactory<>(pulsarClient, + Collections.emptyMap()); + PulsarTemplate template = new PulsarTemplate<>(pulsarProducerFactory); + template.setSchema(ProtobufSchema.of(Proto.Person.class)); + + for (int i = 0; i < 3; i++) { + template.send("protobuf-topic", Proto.Person.newBuilder().setId(i).setName("Paul").build()); + } + assertThat(protobufLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(protobufBatchLatch.await(10, TimeUnit.SECONDS)).isTrue(); } @EnablePulsar @@ -340,18 +370,50 @@ public class PulsarListenerTests extends AbstractContainerBaseTests { jsonLatch.countDown(); } - @PulsarListener(id = "avroListener", topics = "avro-topic", subscriptionName = "subscription-5", + @PulsarListener(id = "jsonBatchListener", topics = "json-topic", subscriptionName = "subscription-5", + schemaType = SchemaType.JSON, batch = true, properties = { "subscriptionInitialPosition=Earliest" }) + void listenJsonBatch(List messages) { + messages.forEach(m -> jsonBatchLatch.countDown()); + } + + @PulsarListener(id = "avroListener", topics = "avro-topic", subscriptionName = "subscription-6", schemaType = SchemaType.AVRO, properties = { "subscriptionInitialPosition=Earliest" }) void listenAvro(User message) { avroLatch.countDown(); } - @PulsarListener(id = "keyvalueListener", topics = "keyvalue-topic", subscriptionName = "subscription-6", + @PulsarListener(id = "avroBatchListener", topics = "avro-topic", subscriptionName = "subscription-7", + schemaType = SchemaType.AVRO, batch = true, properties = { "subscriptionInitialPosition=Earliest" }) + void listenAvroBatch(Messages messages) { + messages.forEach(m -> avroBatchLatch.countDown()); + } + + @PulsarListener(id = "keyvalueListener", topics = "keyvalue-topic", subscriptionName = "subscription-8", schemaType = SchemaType.KEY_VALUE, properties = { "subscriptionInitialPosition=Earliest" }) void listenKeyvalue(KeyValue message) { keyvalueLatch.countDown(); } + @PulsarListener(id = "keyvalueBatchListener", topics = "keyvalue-topic", + subscriptionName = "subscription-9", schemaType = SchemaType.KEY_VALUE, batch = true, + properties = { "subscriptionInitialPosition=Earliest" }) + void listenKeyvalueBatch(List> messages) { + messages.forEach(m -> keyvalueBatchLatch.countDown()); + } + + @PulsarListener(id = "protobufListener", topics = "protobuf-topic", subscriptionName = "subscription-10", + schemaType = SchemaType.PROTOBUF, properties = { "subscriptionInitialPosition=Earliest" }) + void listenProtobuf(Proto.Person message) { + protobufLatch.countDown(); + } + + @PulsarListener(id = "protobufBatchListener", topics = "protobuf-topic", + subscriptionName = "subscription-11", schemaType = SchemaType.PROTOBUF, batch = true, + properties = { "subscriptionInitialPosition=Earliest" }) + void listenProtobufBatch(List messages) { + messages.forEach(m -> protobufBatchLatch.countDown()); + } + } static class User { diff --git a/src/checkstyle/checkstyle-suppressions.xml b/src/checkstyle/checkstyle-suppressions.xml index bb3953ff..552efb75 100644 --- a/src/checkstyle/checkstyle-suppressions.xml +++ b/src/checkstyle/checkstyle-suppressions.xml @@ -7,6 +7,7 @@ +