Fixes GH-219 (#220)

- Default contentType is now application/json. Messages without a contentType will have it set by default (same as docs)
- Fixed issue where manual ack was being verified on the wrong place, causing an warning to be displayed when not needed

gcp-pubsub/src/main/java/io/micronaut/gcp/pubsub/annotation/Subscription.java

@@ -38,10 +38,10 @@
 
     /**
      * Defines the Content-Type to be used for message deserialization.
-     * There's no default, if not set and the message does not contain a Content-Type header deserialization will fail.
+     * Defaults to application/json.
      * @return contentType to use
      */
-    String contentType() default "";
+    String contentType() default "application/json";
 
     /**
      * Defines the name of a particular configuration used for a Subscriber.

gcp-pubsub/src/main/java/io/micronaut/gcp/pubsub/bind/PubSubBodyBinder.java

@@ -18,6 +18,7 @@
 import com.google.pubsub.v1.PubsubMessage;
 import io.micronaut.core.convert.ArgumentConversionContext;
 import io.micronaut.core.type.Argument;
+import io.micronaut.core.util.StringUtils;
 import io.micronaut.gcp.pubsub.exception.PubSubListenerException;
 import io.micronaut.gcp.pubsub.serdes.PubSubMessageSerDes;
 import io.micronaut.gcp.pubsub.serdes.PubSubMessageSerDesRegistry;
@@ -56,7 +57,7 @@ public BindingResult<Object> bind(ArgumentConversionContext<Object> context, Pub
         } else if (bodyType.getType().equals(PubsubMessage.class)) {
             result = state.getPubsubMessage();
         } else {
-            if (!state.getPubsubMessage().containsAttributes("Content-Type")) {
+            if (StringUtils.isEmpty(state.getContentType()) && !state.getPubsubMessage().containsAttributes("Content-Type")) {
                 throw  new PubSubListenerException("Could not detect Content-Type header at message and no Content-Type specified on method.");
             }
             PubSubMessageSerDes serDes = serDesRegistry.find(state.getContentType())

gcp-pubsub/src/main/java/io/micronaut/gcp/pubsub/intercept/PubSubConsumerAdvice.java

@@ -110,14 +110,14 @@ public void process(BeanDefinition<?> beanDefinition, ExecutableMethod<?, ?> met
             ProjectSubscriptionName projectSubscriptionName = PubSubSubscriptionUtils.toProjectSubscriptionName(subscriptionName, googleCloudConfiguration.getProjectId());
             String defaultContentType = subscriptionAnnotation.get("contentType", String.class).orElse("");
             String configuration = subscriptionAnnotation.get("configuration", String.class).orElse("");
-            MessageReceiver receiver = (PubsubMessage message, AckReplyConsumer consumer) -> {
+            MessageReceiver receiver = (PubsubMessage message, AckReplyConsumer ackReplyConsumer) -> {
                 String messageContentType = message.getAttributesMap().getOrDefault("Content-Type", "");
                 String contentType = Optional.of(messageContentType)
                         .filter(StringUtils::isNotEmpty)
                         .orElse(defaultContentType);
-                DefaultPubSubAcknowledgement pubSubAcknowledgement = new DefaultPubSubAcknowledgement(consumer);
+                DefaultPubSubAcknowledgement pubSubAcknowledgement = new DefaultPubSubAcknowledgement(ackReplyConsumer);
 
-                PubSubConsumerState consumerState = new PubSubConsumerState(message, consumer,
+                PubSubConsumerState consumerState = new PubSubConsumerState(message, ackReplyConsumer,
                         projectSubscriptionName, contentType);
                 try {
                     BoundExecutable executable = null;
@@ -130,8 +130,12 @@ public void process(BeanDefinition<?> beanDefinition, ExecutableMethod<?, ?> met
                     if (!hasAckArg) { // if manual ack is not specified we auto ack message after method execution
                         pubSubAcknowledgement.ack();
                     } else {
-                        if (!pubSubAcknowledgement.isClientAck()) {
-                            logger.warn("Method {} was executed and no message acknowledge detected. Did you forget to invoke ack()/nack()?", method.getName());
+                        Optional<Object> boundAck = Arrays.stream(executable.getBoundArguments()).filter(o -> (o instanceof DefaultPubSubAcknowledgement)).findFirst();
+                        if (boundAck.isPresent()) {
+                            DefaultPubSubAcknowledgement manualAck = (DefaultPubSubAcknowledgement) boundAck.get();
+                            if (!manualAck.isClientAck()) {
+                                logger.warn("Method {} was executed and no message acknowledge detected. Did you forget to invoke ack()/nack()?", method.getName());
+                            }
                         }
                     }
                 } catch (Exception e) {

gcp-pubsub/src/test/groovy/io/micronaut/gcp/pubsub/bind/SimpleConsumerSpec.groovy

@@ -6,12 +6,15 @@ import com.google.pubsub.v1.PubsubMessage
 import io.micronaut.context.annotation.Property
 import io.micronaut.context.annotation.Requires
 import io.micronaut.gcp.pubsub.AbstractConsumerSpec
+import io.micronaut.gcp.pubsub.MockPubSubEngine
 import io.micronaut.gcp.pubsub.annotation.MessageId
 import io.micronaut.gcp.pubsub.annotation.PubSubClient
 import io.micronaut.gcp.pubsub.annotation.PubSubListener
 import io.micronaut.gcp.pubsub.annotation.Subscription
 import io.micronaut.gcp.pubsub.annotation.Topic
 import io.micronaut.gcp.pubsub.support.Person
+import io.micronaut.http.annotation.Body
+import io.micronaut.messaging.Acknowledgement
 import io.micronaut.messaging.annotation.Header
 import io.micronaut.test.annotation.MicronautTest
 import spock.util.concurrent.PollingConditions
@@ -31,6 +34,8 @@ class SimpleConsumerSpec extends AbstractConsumerSpec {
     @Inject
     ObjectMapper mapper
 
+    @Inject MockPubSubEngine mockPubSubEngine
+
     void "simple consumer spec"() {
         PollingConditions conditions = new PollingConditions(timeout: 3)
         when:
@@ -93,6 +98,35 @@ class SimpleConsumerSpec extends AbstractConsumerSpec {
                 receiver.dataHolder["test-without-content-type"].name == person.name
             }
     }
+
+    void "receive with manual ack"() {
+        PollingConditions conditions = new PollingConditions(timeout: 3)
+        def person = new Person()
+        person.name = "alf"
+        when:
+            pubSubClient.publishPojoForManualAck(person)
+        then:
+            conditions.eventually {
+                receiver.dataHolder["test-with-manual-ack"].name == person.name
+            }
+    }
+
+    void "receive with default content type used"() {
+        PollingConditions conditions = new PollingConditions(timeout: 3)
+        def person = new Person()
+        person.name = "alf"
+        def message = PubsubMessage
+                .newBuilder()
+                .setData(ByteString.copyFrom(mapper.writeValueAsBytes(person)))
+                .build()
+
+        when:
+            mockPubSubEngine.publish(message, "test-with-default-contentType")
+        then:
+            conditions.eventually {
+                receiver.dataHolder["test-with-default-contentType"].name == person.name
+            }
+    }
 }
 
 @PubSubClient
@@ -113,6 +147,9 @@ interface SimplePubSubClient {
     @Topic(value = "test-without-content-type", contentType = "")
     String publishPojoWithoutContentType(byte[] data)
 
+    @Topic(value = "test-with-manual-ack")
+    String publishPojoForManualAck(Person person);
+
 }
 
 @PubSubListener
@@ -150,4 +187,14 @@ class SimpleReceiver {
     void receive(Person person){
         dataHolder["test-without-content-type"] = person
     }
+
+    @Subscription(value = "test-with-manual-ack", contentType = "application/json")
+    void receive(@Body Person person, Acknowledgement ack) {
+        dataHolder["test-with-manual-ack"] = person
+        ack.ack()
+    }
+    @Subscription(value="test-with-default-contentType")
+    void receiveWithDefault(Person person){
+        dataHolder["test-with-default-contentType"] = person
+    }
 }