/*
 * Copyright 2017-2020 original authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package example.background;

import com.google.cloud.functions.*;
import io.micronaut.gcp.function.GoogleFunctionInitializer;

import jakarta.inject.*;
import java.util.*;

public class Example extends GoogleFunctionInitializer // (1)
        implements BackgroundFunction<PubSubMessage> { // (2)

    @Inject LoggingService loggingService; // (3)

    @Override
    public void accept(PubSubMessage message, Context context) {
        loggingService.logMessage(message);
    }
}

class PubSubMessage {
    String data;
    Map<String, String> attributes;
    String messageId;
    String publishTime;
}

@Singleton
class LoggingService {

    void logMessage(PubSubMessage message) {
        // log the message
    }
}

/*
 * Copyright 2017-2020 original authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package example.background

import com.google.cloud.functions.*
import io.micronaut.gcp.function.GoogleFunctionInitializer

import jakarta.inject.*

class Example extends GoogleFunctionInitializer // (1)
        implements BackgroundFunction<PubSubMessage> { // (2)

    @Inject LoggingService loggingService // (3)

    @Override
    void accept(PubSubMessage message, Context context) {
        loggingService.logMessage(message)
    }
}

class PubSubMessage {
    String data
    Map<String, String> attributes
    String messageId
    String publishTime
}

@Singleton
class LoggingService {

    void logMessage(PubSubMessage message) {
        // log the message
    }
}

/*
 * Copyright 2017-2020 original authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package example.background

import com.google.cloud.functions.BackgroundFunction
import com.google.cloud.functions.Context
import io.micronaut.gcp.function.GoogleFunctionInitializer
import jakarta.inject.Inject
import jakarta.inject.Singleton

class Example : GoogleFunctionInitializer(), // (1)
        BackgroundFunction<PubSubMessage> { // (2)
    @Inject
    lateinit var loggingService: LoggingService // (3)

    override fun accept(message: PubSubMessage, context: Context) {
        loggingService.logMessage(message)
    }
}

class PubSubMessage {
    var data: String? = null
    var attributes: Map<String, String>? = null
    var messageId: String? = null
    var publishTime: String? = null
}

@Singleton
class LoggingService {

    fun logMessage(message: PubSubMessage) {
        // log the message
    }
}

import com.google.pubsub.v1.PubsubMessage;
import io.micronaut.gcp.pubsub.annotation.PubSubClient;
import io.micronaut.gcp.pubsub.annotation.Topic;
import io.micronaut.gcp.pubsub.support.Animal;
import io.micronaut.http.MediaType;

@PubSubClient
public interface CustomSerDesClient {

    @Topic("animals") // (1)
    void send(PubsubMessage pubsubMessage);

    @Topic("animals") // (2)
    void send(byte[] data);

    @Topic(value = "animals", contentType = MediaType.IMAGE_GIF) // (3)
    void sendWithCustomType(byte[] data);

    @Topic("animals") // (4)
    void send(Animal animal);

    @Topic(value = "animals", contentType = MediaType.APPLICATION_XML) // (5)
    void sendWithCustomType(Animal animal);

}

import com.google.pubsub.v1.PubsubMessage;
import io.micronaut.gcp.pubsub.annotation.PubSubClient;
import io.micronaut.gcp.pubsub.annotation.Topic;
import io.micronaut.gcp.pubsub.support.Animal;
import io.micronaut.http.MediaType;

@PubSubClient
interface CustomSerDesClient {

    @Topic("animals") // (1)
    void send(PubsubMessage pubsubMessage)

    @Topic("animals") // (2)
    void send(byte[] data)

    @Topic(value = "animals", contentType = MediaType.IMAGE_GIF) // (3)
    void sendWithCustomType(byte[] data)

    @Topic("animals") // (4)
    void send(Animal animal)

    @Topic(value = "animals", contentType = MediaType.APPLICATION_XML) // (5)
    void sendWithCustomType(Animal animal)

}

import com.google.pubsub.v1.PubsubMessage
import io.micronaut.gcp.pubsub.annotation.PubSubClient
import io.micronaut.gcp.pubsub.annotation.Topic
import io.micronaut.gcp.pubsub.support.Animal
import io.micronaut.http.MediaType

@PubSubClient
interface CustomSerDesClient {

	@Topic("animals") // (1)
	fun send(pubsubMessage: PubsubMessage)

	@Topic("animals")  // (2)
	fun send(data: ByteArray)

	@Topic(value = "animals", contentType = MediaType.IMAGE_GIF) // (3)
	fun sendWithCustomType(data: ByteArray)

	@Topic("animals") // (4)
	fun send(animal: Animal)

	@Topic(value = "animals", contentType = MediaType.APPLICATION_XML) // (5)
	fun sendWithCustomType(animal: Animal)

}

import io.micronaut.gcp.pubsub.annotation.PubSubClient;
import io.micronaut.gcp.pubsub.annotation.Topic;
import io.micronaut.gcp.pubsub.support.Animal;
import io.micronaut.messaging.annotation.MessageHeader;

@PubSubClient
@MessageHeader(name = "application-name", value = "petclinic") // (1)
public interface CustomHeadersClient {

    @MessageHeader(name = "status", value = "healthy") // (2)
    @Topic("animals")
    void sendWithStaticHeaders(Animal animal);

    @Topic("animals")
    void sendWithDynamicHeaders(Animal animal, @MessageHeader(name = "code") Integer code); // (3)

}

import io.micronaut.gcp.pubsub.annotation.PubSubClient;
import io.micronaut.gcp.pubsub.annotation.Topic;
import io.micronaut.gcp.pubsub.support.Animal
import io.micronaut.messaging.annotation.MessageHeader;

@PubSubClient
@MessageHeader(name = "application-name", value = "petclinic") // (1)
 interface CustomHeadersClient {

    @MessageHeader(name = "status", value = "healthy") // (2)
    @Topic("animals")
    void sendWithStaticHeaders(Animal animal)

    @Topic("animals")
    void sendWithDynamicHeaders(Animal animal, @MessageHeader(name = "code") Integer code) // (3)

}

import io.micronaut.gcp.pubsub.annotation.PubSubClient
import io.micronaut.gcp.pubsub.annotation.Topic
import io.micronaut.gcp.pubsub.support.Animal
import io.micronaut.messaging.annotation.MessageHeader


@PubSubClient
@MessageHeader(name = "application-name", value = "petclinic") // (1)
interface CustomHeadersClient {

	@MessageHeader(name = "status", value = "healthy") // (2)
	@Topic("animals")
	fun sendWithStaticHeaders(animal: Animal)

	@Topic("animals")
	fun sendWithDynamicHeaders(animal: Animal, @MessageHeader(name = "code") code: Int) // (3)

}

import io.micronaut.gcp.pubsub.annotation.PubSubClient;
import io.micronaut.gcp.pubsub.annotation.Topic;
import io.micronaut.gcp.pubsub.support.Animal;

@PubSubClient
public interface CustomConfigurationClient {

    @Topic(value = "animals", configuration = "batching") // (1)
    void batchSend(Animal animal);

    @Topic(value = "animals", configuration = "immediate") // (2)
    void send(Animal animal);

}

import io.micronaut.gcp.pubsub.annotation.PubSubClient;
import io.micronaut.gcp.pubsub.annotation.Topic;
import io.micronaut.gcp.pubsub.support.Animal;

@PubSubClient
interface CustomConfigurationClient {

    @Topic(value = "animals", configuration = "batching") // (1)
    void batchSend(Animal animal)

    @Topic(value = "animals", configuration = "immediate") // (2)
    void send(Animal animal)

}

import io.micronaut.gcp.pubsub.annotation.PubSubClient
import io.micronaut.gcp.pubsub.annotation.Topic
import io.micronaut.gcp.pubsub.support.Animal

@PubSubClient
interface CustomConfigurationClient {
	@Topic(value = "animals", configuration = "batching")
	fun batchSend(animal: Animal)  // (1)

	@Topic(value = "animals", configuration = "immediate")
	fun send(animal: Animal)  // (2)
}

import io.micronaut.gcp.pubsub.annotation.PubSubClient;
import io.micronaut.gcp.pubsub.annotation.Topic;
import io.micronaut.gcp.pubsub.support.Animal;
import reactor.core.publisher.Mono;

@PubSubClient
public interface CustomReturnClient {

    @Topic("animals")
    void send(Animal animal); // (1)

    @Topic("animals")
    String sendWithId(Animal animal); // (2)

    @Topic("animals")
    Mono<String> sendReactive(Animal animal); //(3)

}

import io.micronaut.gcp.pubsub.annotation.PubSubClient
import io.micronaut.gcp.pubsub.annotation.Topic
import io.micronaut.gcp.pubsub.support.Animal
import reactor.core.publisher.Mono


@PubSubClient
interface CustomReturnClient {

    @Topic("animals")
    void send(Animal animal) // (1)

    @Topic("animals")
    String sendWithId(Animal animal) // (2)

    @Topic("animals")
    Mono<String> sendReactive(Animal animal) //(3)

}

import io.micronaut.gcp.pubsub.annotation.PubSubClient
import io.micronaut.gcp.pubsub.annotation.Topic
import io.micronaut.gcp.pubsub.support.Animal
import reactor.core.publisher.Mono


@PubSubClient
interface CustomReturnClient {

	@Topic("animals")
	fun send(animal: Animal) // (1)

	@Topic("animals")
	fun sendWithId(animal: Animal): String // (2)

	@Topic("animals")
	fun sendReactive(animal: Animal?): Mono<String> //(3)
}

import com.google.pubsub.v1.PubsubMessage;
import io.micronaut.context.annotation.Requires;
import io.micronaut.gcp.pubsub.annotation.MessageId;
import io.micronaut.gcp.pubsub.annotation.PubSubListener;
import io.micronaut.gcp.pubsub.annotation.Subscription;
import io.micronaut.gcp.pubsub.support.Animal;
import io.micronaut.http.MediaType;

@PubSubListener
public class ContentTypeSubscriber {

    @Subscription("raw-subscription") // (1)
    void receiveRaw(byte[] data, @MessageId String id) {
    }

    @Subscription("native-subscription") // (2)
    void receiveNative(PubsubMessage message) {
    }

    @Subscription("animals") // (3)
    void receivePojo(Animal animal, @MessageId String id) {
    }

    @Subscription(value = "animals-legacy", contentType = MediaType.APPLICATION_XML) // (4)
    void receiveXML(Animal animal, @MessageId String id) {
    }

}

import com.google.pubsub.v1.PubsubMessage
import io.micronaut.context.annotation.Requires;
import io.micronaut.gcp.pubsub.annotation.MessageId;
import io.micronaut.gcp.pubsub.annotation.PubSubListener;
import io.micronaut.gcp.pubsub.annotation.Subscription;
import io.micronaut.gcp.pubsub.support.Animal;

@PubSubListener
class ContentTypeSubscriber {

    @Subscription("raw-subscription") // (1)
    void receiveRaw(byte[] data, @MessageId String id) {
    }

    @Subscription("native-subscription") // (2)
    void receiveNative(PubsubMessage message) {
    }

    @Subscription("animals") // (3)
    void receivePojo(Animal animal, @MessageId String id) {
    }

    @Subscription(value = "animals-legacy", contentType = "application/xml") // (4)
    void receiveXML(Animal animal, @MessageId String id) {
    }

}

import com.google.pubsub.v1.PubsubMessage
import io.micronaut.context.annotation.Requires
import io.micronaut.gcp.pubsub.annotation.MessageId
import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.Subscription
import io.micronaut.gcp.pubsub.support.Animal

@PubSubListener
class ContentTypeSubscriber(private val messageProcessor: MessageProcessor)  {
	@Subscription("raw-subscription")
	fun	receiveRaw(data: ByteArray, @MessageId id: String) { // (1)
        messageProcessor.handleByteArrayMessage(data).block()
	}

	@Subscription("native-subscription")
	fun  receiveNative(message: PubsubMessage) { // (2)
        messageProcessor.handlePubsubMessage(message).block()
	}

	@Subscription("animals")
	fun receivePojo(animal: Animal, @MessageId id: String) { // (3)
        messageProcessor.handleAnimalMessage(animal).block()
	}

	@Subscription(value = "animals-legacy", contentType = "application/xml")
	fun  receiveXML(animal: Animal, @MessageId id: String) { // (4)
        messageProcessor.handleAnimalMessage(animal).block()
	}
}

import com.google.pubsub.v1.PubsubMessage;
import io.micronaut.context.annotation.Requires;
import io.micronaut.gcp.pubsub.annotation.MessageId;
import io.micronaut.gcp.pubsub.annotation.PubSubListener;
import io.micronaut.gcp.pubsub.annotation.PushSubscription;
import io.micronaut.gcp.pubsub.support.Animal;
import io.micronaut.http.MediaType;
import io.micronaut.scheduling.TaskExecutors;
import io.micronaut.scheduling.annotation.ExecuteOn;

@PubSubListener
@ExecuteOn(TaskExecutors.BLOCKING) // (1)
public class ContentTypePushSubscriber {

    @PushSubscription("raw-push-subscription") // (2)
    void receiveRaw(byte[] data, @MessageId String id) {
        //process with blocking code
    }

    @PushSubscription("native-push-subscription") // (3)
    void receiveNative(PubsubMessage message) {
        //process with blocking code
    }

    @PushSubscription("animals-push") // (4)
    void receivePojo(Animal animal, @MessageId String id) {
        //process with blocking code
    }

    @PushSubscription(value = "animals-legacy-push", contentType = MediaType.APPLICATION_XML) // (5)
    void receiveXML(Animal animal, @MessageId String id) {
        //process with blocking code
    }

}

import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.PushSubscription
import io.micronaut.gcp.pubsub.support.Animal
import io.micronaut.scheduling.TaskExecutors
import io.micronaut.scheduling.annotation.ExecuteOn

@PubSubListener
@ExecuteOn(TaskExecutors.BLOCKING) // (1)
class ContentTypePushSubscriber {

    @PushSubscription("raw-push-subscription") // (2)
    void receiveRaw(byte[] data, @MessageId String id) {
        //process with blocking code
    }

    @PushSubscription("native-push-subscription") // (3)
    void receiveNative(PubsubMessage message) {
        //process with blocking code
    }

    @PushSubscription("animals-push") // (4)
    void receivePojo(Animal animal, @MessageId String id) {
        //process with blocking code
    }

    @PushSubscription(value = "animals-legacy-push", contentType = "application/xml") // (5)
    void receiveXML(Animal animal, @MessageId String id) {
        //process with blocking code
    }

}

import com.google.pubsub.v1.PubsubMessage
import io.micronaut.context.annotation.Requires
import io.micronaut.gcp.pubsub.annotation.MessageId
import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.PushSubscription
import io.micronaut.gcp.pubsub.support.Animal
import io.micronaut.scheduling.TaskExecutors
import io.micronaut.scheduling.annotation.ExecuteOn


@PubSubListener
@ExecuteOn(TaskExecutors.BLOCKING) // (1)
class ContentTypePushSubscriber(private val messageProcessor: MessageProcessor)  {
	@PushSubscription("raw-push-subscription")
	fun	receiveRaw(data: ByteArray, @MessageId id: String) { // (2)
        messageProcessor.handleByteArrayMessage(data).block()
	}

	@PushSubscription("native-push-subscription")
	fun  receiveNative(message: PubsubMessage) { // (3)
        messageProcessor.handlePubsubMessage(message).block()
	}

	@PushSubscription("animals-push")
	fun receivePojo(animal: Animal, @MessageId id: String) { // (4)
        messageProcessor.handleAnimalMessage(animal).block()
	}

	@PushSubscription(value = "animals-legacy-push", contentType = "application/xml")
	fun  receiveXML(animal: Animal, @MessageId id: String) { // (5)
        messageProcessor.handleAnimalMessage(animal).block()
	}
}

import com.google.pubsub.v1.PubsubMessage;
import io.micronaut.context.annotation.Requires;
import io.micronaut.gcp.pubsub.annotation.MessageId;
import io.micronaut.gcp.pubsub.annotation.PubSubListener;
import io.micronaut.gcp.pubsub.annotation.Subscription;
import io.micronaut.gcp.pubsub.support.Animal;
import reactor.core.publisher.Mono;

@PubSubListener
public class ReactiveSubscriber {

    private final MessageProcessor messageProcessor;

    public ReactiveSubscriber(MessageProcessor messageProcessor) {
        this.messageProcessor = messageProcessor;
    }

    @Subscription("raw-subscription") // (1)
    Mono<Object> receiveRaw(Mono<byte[]> data, @MessageId String id) {
        return data.flatMap(messageProcessor::handleByteArrayMessage);
    }

    @Subscription("native-subscription") // (2)
    Mono<Object> receiveNative(Mono<PubsubMessage> message) {
        return message.flatMap(messageProcessor::handlePubSubMessage);
    }

    @Subscription("animals") // (3)
    Mono<Object> receivePojo(Mono<Animal> animal, @MessageId String id) {
        return animal.flatMap(messageProcessor::handleAnimalMessage);
    }

    @Subscription(value = "animals-legacy", contentType = "application/xml") // (4)
    Mono<Object> receiveXML(Mono<Animal> animal, @MessageId String id) {
        return animal.flatMap(messageProcessor::handleAnimalMessage);
    }
}

import io.micronaut.gcp.pubsub.annotation.MessageId
import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.Subscription
import io.micronaut.gcp.pubsub.support.Animal
import reactor.core.publisher.Mono


@PubSubListener
class ReactiveSubscriber {

    private final MessageProcessor messageProcessor

    ReactiveSubscriber(MessageProcessor messageProcessor) {
        this.messageProcessor = messageProcessor
    }

    @Subscription("raw-subscription") // (1)
    Mono<Object> receiveRaw(Mono<byte[]> data, @MessageId String id) {
        return data.flatMap(messageProcessor::handleByteArrayMessage)
    }

    @Subscription("native-subscription") // (2)
    Mono<Object> receiveNative(Mono<PubsubMessage> message) {
        return message.flatMap(messageProcessor::handlePubSubMessage)
    }

    @Subscription("animals") // (3)
    Mono<Object> receivePojo(Mono<Animal> animal, @MessageId String id) {
        return animal.flatMap(messageProcessor::handleAnimalMessage)
    }

    @Subscription(value = "animals-legacy", contentType = "application/xml") // (4)
    Mono<Object> receiveXML(Mono<Animal> animal, @MessageId String id) {
        return animal.flatMap(messageProcessor::handleAnimalMessage)
    }

}

import com.google.pubsub.v1.PubsubMessage
import io.micronaut.context.annotation.Requires
import io.micronaut.gcp.pubsub.annotation.MessageId
import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.Subscription
import io.micronaut.gcp.pubsub.support.Animal
import reactor.core.publisher.Mono


@PubSubListener
class ReactiveSubscriber(private val messageProcessor: MessageProcessor) {

    @Subscription("raw-subscription")
    fun receiveRaw(data: Mono<ByteArray>, @MessageId id: String): Mono<Any> { // (1)
        return data.flatMap { payload ->
            messageProcessor.handleByteArrayMessage(payload)
        }
    }

    @Subscription("native-subscription")
    fun receiveNative(message: Mono<PubsubMessage>): Mono<Any> { // (2)
        return message.flatMap { payload ->
            messageProcessor.handlePubsubMessage(payload)
        }
    }

    @Subscription("animals")
    fun receivePojo(message: Mono<Animal>, @MessageId id: String): Mono<Any> { // (3)
        return message.flatMap { animal ->
            messageProcessor.handleAnimalMessage(animal)
        }
    }

    @Subscription(value = "animals-legacy", contentType = "application/xml")
    fun receiveXML(message: Mono<Animal>, @MessageId id: String): Mono<Any> { // (4)
        return message.flatMap { animal ->
            messageProcessor.handleAnimalMessage(animal)
        }
    }
}

import com.google.pubsub.v1.PubsubMessage;
import io.micronaut.context.annotation.Requires;
import io.micronaut.gcp.pubsub.annotation.MessageId;
import io.micronaut.gcp.pubsub.annotation.PubSubListener;
import io.micronaut.gcp.pubsub.annotation.PushSubscription;
import io.micronaut.gcp.pubsub.support.Animal;
import reactor.core.publisher.Mono;

@PubSubListener
public class ReactivePushSubscriber {

    private final MessageProcessor messageProcessor;

    public ReactivePushSubscriber(MessageProcessor messageProcessor) {
        this.messageProcessor = messageProcessor;
    }

    @PushSubscription("raw-push-subscription") // (1)
    Mono<Object> receiveRaw(Mono<byte[]> data, @MessageId String id) {
        return data.flatMap(messageProcessor::handleByteArrayMessage);
    }

    @PushSubscription("native-push-subscription") // (2)
    Mono<Object> receiveNative(Mono<PubsubMessage> message) {
        return message.flatMap(messageProcessor::handlePubSubMessage);
    }

    @PushSubscription("animals-push") // (3)
    Mono<Object> receivePojo(Mono<Animal> animal, @MessageId String id) {
        return animal.flatMap(messageProcessor::handleAnimalMessage);
    }

    @PushSubscription(value = "animals-legacy-push", contentType = "application/xml") // (4)
    Mono<Object> receiveXML(Mono<Animal> animal, @MessageId String id) {
        return animal.flatMap(messageProcessor::handleAnimalMessage);
    }
}

import io.micronaut.gcp.pubsub.annotation.PushSubscription
import io.micronaut.gcp.pubsub.support.Animal
import reactor.core.publisher.Mono


@PubSubListener
class ReactivePushSubscriber {

    private final MessageProcessor messageProcessor

    ReactivePushSubscriber(MessageProcessor messageProcessor) {
        this.messageProcessor = messageProcessor
    }

    @PushSubscription("raw-push-subscription") // (1)
    Mono<Object> receiveRaw(Mono<byte[]> data, @MessageId String id) {
        return data.flatMap(messageProcessor::handleByteArrayMessage)
    }

    @PushSubscription("native-push-subscription") // (2)
    Mono<Object> receiveNative(Mono<PubsubMessage> message) {
        return message.flatMap(messageProcessor::handlePubSubMessage)
    }

    @PushSubscription("animals-push") // (3)
    Mono<Object> receivePojo(Mono<Animal> animal, @MessageId String id) {
        return animal.flatMap(messageProcessor::handleAnimalMessage)
    }

    @PushSubscription(value = "animals-legacy-push", contentType = "application/xml") // (4)
    Mono<Object> receiveXML(Mono<Animal> animal, @MessageId String id) {
        return animal.flatMap(messageProcessor::handleAnimalMessage)
    }

}

import com.google.pubsub.v1.PubsubMessage
import io.micronaut.context.annotation.Requires
import io.micronaut.gcp.pubsub.annotation.MessageId
import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.PushSubscription
import io.micronaut.gcp.pubsub.support.Animal
import reactor.core.publisher.Mono


@PubSubListener
class ReactivePushSubscriber(private val messageProcessor: MessageProcessor) {

    @PushSubscription("raw-push-subscription")
    fun receiveRaw(data: Mono<ByteArray>, @MessageId id: String): Mono<Any> { // (1)
        return data.flatMap { payload ->
            messageProcessor.handleByteArrayMessage(payload)
        }
    }

    @PushSubscription("native-push-subscription")
    fun receiveNative(message: Mono<PubsubMessage>): Mono<Any> { // (2)
        return message.flatMap { payload ->
            messageProcessor.handlePubsubMessage(payload)
        }
    }

    @PushSubscription("animals-push")
    fun receivePojo(message: Mono<Animal>, @MessageId id: String): Mono<Any> { // (3)
        return message.flatMap { animal ->
            messageProcessor.handleAnimalMessage(animal)
        }
    }

    @PushSubscription(value = "animals-legacy-push", contentType = "application/xml")
    fun receiveXML(message: Mono<Animal>, @MessageId id: String): Mono<Any> { // (4)
        return message.flatMap { animal ->
            messageProcessor.handleAnimalMessage(animal)
        }
    }
}

import io.micronaut.context.annotation.Requires;
import io.micronaut.gcp.pubsub.annotation.PubSubListener;
import io.micronaut.gcp.pubsub.annotation.Subscription;
import io.micronaut.gcp.pubsub.support.Animal;
import io.micronaut.messaging.annotation.MessageHeader;

@PubSubListener
public class CustomHeaderSubscriber {

    @Subscription("animals")
    public void onMessage(Animal animal, @MessageHeader("Content-Type") String contentType, @MessageHeader("code") Integer code) { // (1)

    }

}

import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.Subscription
import io.micronaut.gcp.pubsub.support.Animal
import io.micronaut.messaging.annotation.MessageHeader

@PubSubListener
class CustomHeaderSubscriber {

    @Subscription("animals")
    void onMessage(Animal animal, @MessageHeader("Content-Type") String contentType, @MessageHeader("code") Integer code) { // (1)

    }

}

import io.micronaut.context.annotation.Requires
import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.Subscription
import io.micronaut.gcp.pubsub.support.Animal
import io.micronaut.messaging.annotation.MessageHeader

@PubSubListener
class CustomHeaderSubscriber {

	@Subscription("animals")
	fun onMessage(animal: Animal,
				  @MessageHeader("Content-Type") contentType: String, @MessageHeader("code") code: Int) { // (1)
	}
}

import io.micronaut.context.annotation.Requires;
import io.micronaut.gcp.pubsub.annotation.PubSubListener;
import io.micronaut.gcp.pubsub.annotation.Subscription;
import io.micronaut.gcp.pubsub.support.Animal;

@PubSubListener
public class CustomConfigurationSubscriber {

    @Subscription(value = "animals", configuration = "custom") // (1)
    public void onMessage(Animal animal) {

    }

}

import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.Subscription
import io.micronaut.gcp.pubsub.support.Animal

@PubSubListener
class CustomConfigurationSubscriber {

    @Subscription(value = "animals", configuration = "custom") // (1)
    void onMessage(Animal animal) {

    }

}

import io.micronaut.context.annotation.Requires
import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.Subscription
import io.micronaut.gcp.pubsub.support.Animal

@PubSubListener
class CustomConfigurationSubscriber {

	@Subscription(value = "animals", configuration = "custom") // (1)
	fun onMessage(animal: Animal) {

	}

}

import io.micronaut.context.annotation.Requires;
import io.micronaut.gcp.pubsub.annotation.PubSubListener;
import io.micronaut.gcp.pubsub.annotation.Subscription;
import io.micronaut.gcp.pubsub.support.Animal;
import io.micronaut.messaging.Acknowledgement;
import reactor.core.publisher.Mono;

@PubSubListener
public class AcknowledgementSubscriber {

    private final MessageProcessor messageProcessor;

    public AcknowledgementSubscriber(MessageProcessor messageProcessor) {
        this.messageProcessor = messageProcessor;
    }

    @Subscription("animals")
    public void onMessage(Animal animal, Acknowledgement acknowledgement) {
        if (Boolean.TRUE.equals(messageProcessor.handleAnimalMessage(animal).block())) {
            acknowledgement.ack();
        } else {
            acknowledgement.nack();
        }
    }

    @Subscription("animals-async")
    public Mono<Boolean> onReactiveMessage(Mono<Animal> animal, Acknowledgement acknowledgement) {
        return animal.flatMap(messageProcessor::handleAnimalMessage)
            .doOnNext(result -> {
                if (Boolean.TRUE.equals(result)) {
                    acknowledgement.ack();
                } else {
                    acknowledgement.nack();
                }
            });
    }

}

import io.micronaut.gcp.pubsub.annotation.Subscription
import io.micronaut.gcp.pubsub.support.Animal
import io.micronaut.messaging.Acknowledgement
import reactor.core.publisher.Mono


@PubSubListener
class AcknowledgementSubscriber {

    MessageProcessor messageProcessor

    AcknowledgementSubscriber(MessageProcessor messageProcessor) {
        this.messageProcessor = messageProcessor
    }

    @Subscription("animals")
    void onMessage(Animal animal, Acknowledgement acknowledgement) {
        if (Boolean.TRUE == messageProcessor.handleAnimalMessage(animal).block()) {
            acknowledgement.ack()
        } else {
            acknowledgement.nack()
        }
    }

    @Subscription("animals-async")
    Mono<Boolean> onReactiveMessage(Mono<Animal> animal, Acknowledgement acknowledgement) {
        return animal.flatMap(messageProcessor::handleAnimalMessage)
                .doOnNext(result -> {
                    if (Boolean.TRUE == result) {
                        acknowledgement.ack()
                    } else {
                        acknowledgement.nack()
                    }
                })
    }

}

import io.micronaut.context.annotation.Requires
import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.Subscription
import io.micronaut.gcp.pubsub.support.Animal
import io.micronaut.messaging.Acknowledgement
import reactor.core.publisher.Mono


@PubSubListener
class AcknowledgementSubscriber(private val messageProcessor: MessageProcessor) {

    @Subscription("animals")
    fun onMessage(animal: Animal, acknowledgement: Acknowledgement) {
        if (messageProcessor.handleAnimalMessage(animal).block() == true) {
            acknowledgement.ack()
            messageProcessor.recordAcknowledgement(acknowledgement)
        } else {
            acknowledgement.nack()
            messageProcessor.recordAcknowledgement(acknowledgement)
        }
    }

    @Subscription("animals-async")
    fun onReactiveMessage(message: Mono<Animal>, acknowledgement: Acknowledgement): Mono<Boolean> {
        return message.flatMap { animal -> messageProcessor.handleAnimalMessage(animal) }
            .doOnNext { result ->
                if (result) {
                    acknowledgement.ack()
                    messageProcessor.recordAcknowledgement(acknowledgement)
                } else {
                    acknowledgement.nack()
                    messageProcessor.recordAcknowledgement(acknowledgement)
                }
            }
    }
}

import io.micronaut.context.annotation.Requires;
import io.micronaut.gcp.pubsub.annotation.PubSubListener;
import io.micronaut.gcp.pubsub.annotation.PushSubscription;
import io.micronaut.gcp.pubsub.support.Animal;
import io.micronaut.messaging.Acknowledgement;
import io.micronaut.scheduling.TaskExecutors;
import io.micronaut.scheduling.annotation.ExecuteOn;
import reactor.core.publisher.Mono;

@PubSubListener
public class AcknowledgementPushSubscriber {

    private final MessageProcessor messageProcessor;

    public AcknowledgementPushSubscriber(MessageProcessor messageProcessor) {
        this.messageProcessor = messageProcessor;
    }

    @ExecuteOn(TaskExecutors.BLOCKING)
    @PushSubscription("animals-push")
    public void onMessage(Animal animal, Acknowledgement acknowledgement) {
        if (Boolean.TRUE.equals(messageProcessor.handleAnimalMessage(animal).block())) {
            acknowledgement.ack();
        } else {
            acknowledgement.nack();
        }
    }

    @PushSubscription("animals-async-push")
    public Mono<Boolean> onReactiveMessage(Mono<Animal> animal, Acknowledgement acknowledgement) {
        return animal.flatMap(messageProcessor::handleAnimalMessage)
            .doOnNext(result -> {
                if (Boolean.TRUE.equals(result)) {
                    acknowledgement.ack();
                } else {
                    acknowledgement.nack();
                }
            });
    }

}

import io.micronaut.scheduling.TaskExecutors
import io.micronaut.scheduling.annotation.ExecuteOn
import reactor.core.publisher.Mono

@PubSubListener
class AcknowledgementPushSubscriber {

    MessageProcessor messageProcessor

    AcknowledgementPushSubscriber(MessageProcessor messageProcessor) {
        this.messageProcessor = messageProcessor
    }

    @ExecuteOn(TaskExecutors.BLOCKING)
    @PushSubscription("animals-push")
    void onMessage(Animal animal, Acknowledgement acknowledgement) {
        if (Boolean.TRUE == messageProcessor.handleAnimalMessage(animal).block()) {
            acknowledgement.ack()
        } else {
            acknowledgement.nack()
        }
    }

    @PushSubscription("animals-async-push")
    Mono<Boolean> onReactiveMessage(Mono<Animal> animal, Acknowledgement acknowledgement) {
        return animal.flatMap(messageProcessor::handleAnimalMessage)
                .doOnNext(result -> {
                    if (Boolean.TRUE == result) {
                        acknowledgement.ack()
                    } else {
                        acknowledgement.nack()
                    }
                })
    }

}

import io.micronaut.context.annotation.Requires
import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.annotation.PushSubscription
import io.micronaut.gcp.pubsub.support.Animal
import io.micronaut.messaging.Acknowledgement
import io.micronaut.scheduling.TaskExecutors
import io.micronaut.scheduling.annotation.ExecuteOn
import reactor.core.publisher.Mono


@PubSubListener
class AcknowledgementPushSubscriber(private val messageProcessor: MessageProcessor) {

    @ExecuteOn(TaskExecutors.BLOCKING)
    @PushSubscription("animals-push")
    fun onMessage(animal: Animal, acknowledgement: Acknowledgement) {
        if (messageProcessor.handleAnimalMessage(animal).block() == true) {
            acknowledgement.ack()
            messageProcessor.recordAcknowledgement(acknowledgement)
        } else {
            acknowledgement.nack()
            messageProcessor.recordAcknowledgement(acknowledgement)
        }
    }

    @PushSubscription("animals-async-push")
    fun onReactiveMessage(message: Mono<Animal>, acknowledgement: Acknowledgement): Mono<Boolean> {
        return message.flatMap { animal -> messageProcessor.handleAnimalMessage(animal) }
            .doOnNext { result ->
                if (result) {
                    acknowledgement.ack()
                    messageProcessor.recordAcknowledgement(acknowledgement)
                } else {
                    acknowledgement.nack()
                    messageProcessor.recordAcknowledgement(acknowledgement)
                }
            }
    }
}

import com.google.pubsub.v1.PubsubMessage;
import io.micronaut.context.annotation.Requires;
import io.micronaut.gcp.pubsub.annotation.PubSubListener;
import io.micronaut.gcp.pubsub.bind.PubSubConsumerState;
import io.micronaut.gcp.pubsub.exception.PubSubMessageReceiverException;
import io.micronaut.gcp.pubsub.exception.PubSubMessageReceiverExceptionHandler;
import io.micronaut.gcp.pubsub.support.Animal;

@PubSubListener
public class ErrorHandlingSubscriber implements PubSubMessageReceiverExceptionHandler { // (1)
    /**
     *
     * @param animal payload
     */
    public void onMessage(Animal animal) {
        throw new RuntimeException("error");
    }

    @Override
    public void handle(PubSubMessageReceiverException exception) { // (2)

        Object listener = exception.getListener(); // (3)
        PubSubConsumerState state = exception.getState(); // (4)
        PubsubMessage originalMessage = state.getPubsubMessage();
        String contentType = state.getContentType();
        //some logic
        state.getAckReplyConsumer().ack(); // (5)

    }
}

import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.exception.PubSubMessageReceiverException
import io.micronaut.gcp.pubsub.exception.PubSubMessageReceiverExceptionHandler
import io.micronaut.gcp.pubsub.support.Animal

@PubSubListener
class ErrorHandlingSubscriber implements PubSubMessageReceiverExceptionHandler { // (1)

    void onMessage(Animal animal) {
        throw new RuntimeException("error");
    }

    @Override
    void handle(PubSubMessageReceiverException exception) { // (2)

        def listener = exception.listener// (3)
        def state = exception.state // (4)
        def originalMessage = state.pubsubMessage
        def contentType = state.contentType
        //some logic
        state.getAckReplyConsumer().ack(); // (5)

    }
}

import io.micronaut.context.annotation.Requires
import io.micronaut.gcp.pubsub.annotation.PubSubListener
import io.micronaut.gcp.pubsub.exception.PubSubMessageReceiverException
import io.micronaut.gcp.pubsub.exception.PubSubMessageReceiverExceptionHandler
import io.micronaut.gcp.pubsub.support.Animal

@PubSubListener
class ErrorHandlingSubscriber : PubSubMessageReceiverExceptionHandler { // (1)

	fun onMessage(animal: Animal) {
		throw RuntimeException("error")
	}

	override fun handle(exception: PubSubMessageReceiverException) { // (2)

		val listener = exception.listener // (3)
		val state = exception.state // (4)
		val originalMessage = state.pubsubMessage
		val contentType = state.contentType
		//some logic
		state.ackReplyConsumer.ack() // (5)

	}
}

import io.micronaut.core.serialize.exceptions.SerializationException;
import io.micronaut.core.type.Argument;

import jakarta.inject.Singleton;
import java.io.*;

@Singleton // (1)
public class JavaMessageSerDes implements PubSubMessageSerDes {

    @Override
    public String supportedType() { // (2)
        return "application/x.java";
    }

    @Override
    public Object deserialize(byte[] data, Argument<?> type) {
        ByteArrayInputStream bin = new ByteArrayInputStream(data);
        Object result = null;
        try {
            ObjectInputStream reader = new ObjectInputStream(bin);
            result = reader.readObject();
        } catch (Exception e) {
            throw new SerializationException("Failed to deserialize object", e);
        }
        return result;
    }

    @Override
    public byte[] serialize(Object data) {
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        try {
            ObjectOutputStream writer = new ObjectOutputStream(baos);
            writer.writeObject(data);
        } catch (IOException e) {
            throw new SerializationException("Failed to serialize object", e);
        }
        return baos.toByteArray();
    }

}

import io.micronaut.core.serialize.exceptions.SerializationException;
import io.micronaut.core.type.Argument;

import jakarta.inject.Singleton;
import java.io.*;

@Singleton // (1)
class JavaMessageSerDes implements PubSubMessageSerDes {

    @Override
    String supportedType() { // (2)
        "application/x.java"
    }

    @Override
    public Object deserialize(byte[] data, Argument<?> type) {
        ByteArrayInputStream bin = new ByteArrayInputStream(data)
        Object result = null
        try {
            ObjectInputStream reader = new ObjectInputStream(bin)
            result = reader.readObject();
        } catch (Exception e) {
            throw new SerializationException("Failed to deserialize object", e)
        }
        return result;
    }

    @Override
    byte[] serialize(Object data) {
        ByteArrayOutputStream baos = new ByteArrayOutputStream()
        try {
            ObjectOutputStream writer = new ObjectOutputStream(baos)
            writer.writeObject(data)
        } catch (IOException e) {
            throw new SerializationException("Failed to serialize object", e)
        }
        return baos.toByteArray()
    }

}

import io.micronaut.core.serialize.exceptions.SerializationException
import io.micronaut.core.type.Argument
import java.io.*
import jakarta.inject.Singleton

@Singleton // (1)
class JavaMessageSerDes : PubSubMessageSerDes {

	override fun supportedType(): String { // (2)
		return "application/x.java"
	}

	override fun deserialize(data: ByteArray, type: Argument<*>): Any {
		val bin = ByteArrayInputStream(data)
		var result: Any? = null
		result = try {
			val reader = ObjectInputStream(bin)
			reader.readObject()
		} catch (e: Exception) {
			throw SerializationException("Failed to deserialize object", e)
		}
		return result
	}

	override fun serialize(data: Any): ByteArray {
		val baos = ByteArrayOutputStream()
		try {
			val writer = ObjectOutputStream(baos)
			writer.writeObject(data)
		} catch (e: IOException) {
			throw SerializationException("Failed to serialize object", e)
		}
		return baos.toByteArray()
	}
}

@MicronautTest
@Property(name = "spec.name", value = "ContentTypePushSubscriberTest")
@Property(name = "gcp.projectId", value = "test-project")
class ContentTypePushSubscriberSpec {
    @Inject
    JsonMapper jsonMapper;


    @Test
    void testJsonPojo() throws IOException {
        Animal dog = new Animal("dog");

        String encodedData = Base64.getEncoder().encodeToString(jsonMapper.writeValueAsBytes(dog)); // (1)

        PushRequest request = new PushRequest("projects/test-project/subscriptions/animals-push", // (2)
            new PushRequest.PushMessage(new HashMap<>(), encodedData, "1", "2021-02-26T19:13:55.749Z"));

        HttpResponse<?> response = pushClient.toBlocking().exchange(HttpRequest.POST("/push", request)); // (3)

        Assertions.assertEquals(HttpStatus.OK, response.getStatus());

    }

}

@MicronautTest
@Property(name = "spec.name", value = "ContentTypePushSubscriberSpec")
@Property(name = "gcp.projectId", value = "test-project")
class ContentTypePushSubscriberSpec extends Specification {
    @Inject
    JsonMapper jsonMapper


    void "receive pojo message from json"() {
        given:
        Animal dog = new Animal("dog")

        String encodedData = Base64.getEncoder().encodeToString(jsonMapper.writeValueAsBytes(dog)) // (1)

        PushRequest request = new PushRequest("projects/test-project/subscriptions/animals-push", // (2)
                new PushRequest.PushMessage(new HashMap<>(), encodedData, "1", "2021-02-26T19:13:55.749Z"))

        when:
        HttpResponse response = pushClient.toBlocking().exchange(HttpRequest.POST("/push", request)) // (3)

        then:
        response.status == HttpStatus.OK

    }

}