/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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
 *
 *      http://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 org.apache.camel.kafkaconnector.google.pubsub.clients;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import com.google.api.gax.core.NoCredentialsProvider;
import com.google.api.gax.grpc.GrpcTransportChannel;
import com.google.api.gax.rpc.FixedTransportChannelProvider;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.cloud.pubsub.v1.SubscriptionAdminSettings;
import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.cloud.pubsub.v1.TopicAdminSettings;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PushConfig;
import com.google.pubsub.v1.Subscription;
import com.google.pubsub.v1.TopicName;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class GooglePubEasy {
    private static final Logger LOG = LoggerFactory.getLogger(GooglePubEasy.class);
    private final List<String> receivedMessages = new ArrayList<>();

    private final String serviceAddress;
    private final String project;


    private final ManagedChannel channel;
    private final FixedTransportChannelProvider channelProvider;

    private ProjectSubscriptionName projectSubscriptionName;
    private Subscriber subscriber;

    public GooglePubEasy(String serviceAddress, String project) {
        this.serviceAddress = serviceAddress;
        this.project = project;

        channel = ManagedChannelBuilder
                .forTarget(String.format(serviceAddress))
                .usePlaintext()
                .build();

        channelProvider =
                FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel));
    }

    public void createTopic(String topicName) throws IOException, InterruptedException {
        doCreateTopic(topicName);
    }

    public void createSubscription(String subscriptionName, String topicName) throws IOException {
        TopicName googleTopic = TopicName.of(project, topicName);

        projectSubscriptionName = ProjectSubscriptionName.of(project, subscriptionName);

        SubscriptionAdminSettings adminSettings = SubscriptionAdminSettings
                .newBuilder()
                .setCredentialsProvider(NoCredentialsProvider.create())
                .setTransportChannelProvider(channelProvider)
                .build();

        try (SubscriptionAdminClient subscriptionAdminClient = SubscriptionAdminClient.create(adminSettings)) {
            Subscription subscription = subscriptionAdminClient.createSubscription(
                    projectSubscriptionName, googleTopic, PushConfig.getDefaultInstance(), 10);
        }
    }

    private void doCreateTopic(String topicName) throws IOException, InterruptedException {
        TopicName googleTopic = TopicName.of(project, topicName);

        TopicAdminSettings topicAdminSettings = TopicAdminSettings
                .newBuilder()
                .setCredentialsProvider(NoCredentialsProvider.create())
                .setTransportChannelProvider(channelProvider)
                .build();

        try (TopicAdminClient client = TopicAdminClient.create(topicAdminSettings)) {
            LOG.info("Creating topic {} (original {})", googleTopic.toString(), googleTopic.getTopic());

            client.createTopic(googleTopic);

            if (client.awaitTermination(10, TimeUnit.SECONDS)) {
                client.shutdownNow();
            }
        }
    }

    public void receive() {
        try {
            MessageReceiver receiver = (pubsubMessage, ackReplyConsumer) -> {
                String data = pubsubMessage.getData().toString();
                LOG.info("Received: {}", data);
                receivedMessages.add(data);

                if (receivedMessages.size() >= 10) {
                    subscriber.stopAsync();
                }

                ackReplyConsumer.ack();
            };

            subscriber = Subscriber
                    .newBuilder(projectSubscriptionName, receiver)
                    .setCredentialsProvider(NoCredentialsProvider.create())
                    .setChannelProvider(channelProvider)
                    .build();


            LOG.info("Adding listener ...");
            subscriber.addListener(
                    new Subscriber.Listener() {
                        @Override
                        public void failed(Subscriber.State from, Throwable failure) {
                            LOG.error(failure.getMessage(), failure);
                        }
                    },
                    MoreExecutors.directExecutor());

            LOG.info("Starting async ...");
            subscriber.startAsync().awaitRunning();
            LOG.info("Waiting for messages ...");
            subscriber.awaitTerminated(25, TimeUnit.SECONDS);
        } catch (TimeoutException e) {
            subscriber.stopAsync();
        } finally {
            if (subscriber != null) {
                subscriber.stopAsync();
            }
        }
    }

    public void shutdown() {
        if (channel != null) {
            channel.shutdown();
        }
    }

    public List<String> getReceivedMessages() {
        return Collections.unmodifiableList(receivedMessages);
    }
}
