import com.google.cloud.pubsublite.*;
import com.google.cloud.pubsublite.proto.Topic;
import com.google.cloud.pubsublite.proto.Topic.*;
import com.google.protobuf.util.Durations;

// TODO(developer): Replace these variables with your own.
long projectNumber = 123L;
String cloudRegion = "us-central1";
char zoneId = 'b';
String topicId = "your-topic-id";
Integer partitions = 1;

TopicPath topicPath =
    TopicPath.newBuilder()
        .setProject(ProjectNumber.of(projectNumber))
        .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
        .setName(TopicName.of(topicId))
        .build();

Topic topic =
    Topic.newBuilder()
        .setPartitionConfig(
            PartitionConfig.newBuilder()
                // Set publishing throughput to 1 times the standard partition
                // throughput of 4 MiB per sec. This must be in the range [1,4]. A
                // topic with `scale` of 2 and count of 10 is charged for 20 partitions.
                .setScale(1)
                .setCount(partitions))
        .setRetentionConfig(
            RetentionConfig.newBuilder()
                // How long messages are retained.
                .setPeriod(Durations.fromDays(1))
                // Set storage per partition to 30 GiB. This must be 30 GiB-10 TiB.
                // If the number of bytes stored in any of the topic's partitions grows
                // beyond this value, older messages will be dropped to make room for
                // newer ones, regardless of the value of `period`.
                .setPerPartitionBytes(30 * 1024 * 1024 * 1024L))
        .setName(topicPath.toString())
        .build();

AdminClientSettings adminClientSettings =
    AdminClientSettings.newBuilder().setRegion(CloudRegion.of(cloudRegion)).build();

try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
  Topic response = adminClient.createTopic(topic).get();
  System.out.println(response.getAllFields() + "created successfully.");
}

import com.google.api.core.*;
import com.google.cloud.pubsublite.*;
import com.google.cloud.pubsublite.cloudpubsub.*;
import com.google.protobuf.ByteString;
import com.google.pubsub.v1.PubsubMessage;
import java.util.*;

// TODO(developer): Replace these variables before running the sample.
long projectNumber = 123L;
String cloudRegion = "us-central1";
char zoneId = 'b';
// Choose an existing topic.
String topicId = "your-topic-id";
int messageCount = 100;

TopicPath topicPath =
    TopicPath.newBuilder()
        .setProject(ProjectNumber.of(projectNumber))
        .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
        .setName(TopicName.of(topicId))
        .build();
Publisher publisher = null;
List<ApiFuture<String>> futures = new ArrayList<>();

try {
  PublisherSettings publisherSettings =
      PublisherSettings.newBuilder().setTopicPath(topicPath).build();

  publisher = Publisher.create(publisherSettings);

  // Start the publisher. Upon successful starting, its state will become RUNNING.
  publisher.startAsync().awaitRunning();

  for (int i = 0; i < messageCount; i++) {
    String message = "message-" + i;

    // Convert the message to a byte string.
    ByteString data = ByteString.copyFromUtf8(message);
    PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();

    // Publish a message. Messages are automatically batched.
    ApiFuture<String> future = publisher.publish(pubsubMessage);
    futures.add(future);
  }
} finally {
  ArrayList<MessageMetadata> metadata = new ArrayList<>();
  List<String> ackIds = ApiFutures.allAsList(futures).get();
  for (String id : ackIds) {
    // Decoded metadata contains partition and offset.
    metadata.add(MessageMetadata.decode(id));
  }
  System.out.println(metadata + "\nPublished " + ackIds.size() + " messages.");

  if (publisher != null) {
    // Shut down the publisher.
    publisher.stopAsync().awaitTerminated();
    System.out.println("Publisher is shut down.");
  }
}

import com.google.cloud.pubsublite.*;
import com.google.cloud.pubsublite.proto.Subscription;
import com.google.cloud.pubsublite.proto.Subscription.*;
import com.google.cloud.pubsublite.proto.Subscription.DeliveryConfig.*;

// TODO(developer): Replace these variables with your own.
long projectNumber = 123L;
String cloudRegion = "us-central1";
char zoneId = 'b';
// Choose an existing topic.
String topicId = "your-topic-id";
String subscriptionId = "your-subscription-id";

TopicPath topicPath =
    TopicPath.newBuilder()
        .setProject(ProjectNumber.of(projectNumber))
        .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
        .setName(TopicName.of(topicId))
        .build();

SubscriptionPath subscriptionPath =
    SubscriptionPath.newBuilder()
        .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
        .setProject(ProjectNumber.of(projectNumber))
        .setName(SubscriptionName.of(subscriptionId))
        .build();

Subscription subscription =
    Subscription.newBuilder()
        .setDeliveryConfig(
            // The server does not wait for a published message to be successfully
            // written to storage before delivering it to subscribers. As such, a
            // subscriber may receive a message for which the write to storage failed.
            // If the subscriber re-reads the offset of that message later on, there
            // may be a gap at that offset.
            DeliveryConfig.newBuilder()
                .setDeliveryRequirement(DeliveryRequirement.DELIVER_IMMEDIATELY))
        .setName(subscriptionPath.toString())
        .setTopic(topicPath.toString())
        .build();

AdminClientSettings adminClientSettings =
    AdminClientSettings.newBuilder().setRegion(CloudRegion.of(cloudRegion)).build();

try (AdminClient adminClient = AdminClient.create(adminClientSettings)) {
  Subscription response = adminClient.createSubscription(subscription).get();
  System.out.println(response.getAllFields() + "created successfully.");
}

import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsublite.*;
import com.google.cloud.pubsublite.cloudpubsub.*;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.pubsub.v1.PubsubMessage;
import java.util.*;

// TODO(developer): Replace these variables with your own.
long projectNumber = 123L;
String cloudRegion = "us-central1";
char zoneId = 'b';
// Choose an existing topic.
String topicId = "your-topic-id";
// Choose an existing subscription.
String subscriptionId = "your-subscription-id";

SubscriptionPath subscriptionPath =
    SubscriptionPath.newBuilder()
        .setLocation(CloudZone.of(CloudRegion.of(cloudRegion), zoneId))
        .setProject(ProjectNumber.of(projectNumber))
        .setName(SubscriptionName.of(subscriptionId))
        .build();

// The message stream is paused based on the maximum size or number of messages that the
// subscriber has already received, whichever condition is met first.
FlowControlSettings flowControlSettings =
    FlowControlSettings.builder()
        // 10 MiB. Must be greater than the allowed size of the largest message (1 MiB).
        .setBytesOutstanding(10 * 1024 * 1024L)
        // 1,000 outstanding messages. Must be >0.
        .setMessagesOutstanding(1000L)
        .build();

MessageReceiver receiver =
    (PubsubMessage message, AckReplyConsumer consumer) -> {
      System.out.println("Id : " + message.getMessageId());
      System.out.println("Data : " + message.getData().toStringUtf8());
      consumer.ack();
    };

SubscriberSettings subscriberSettings =
    SubscriberSettings.newBuilder()
        .setSubscriptionPath(subscriptionPath)
        .setReceiver(receiver)
        // Flow control settings are set at the partition level.
        .setPerPartitionFlowControlSettings(flowControlSettings)
        .build();

Subscriber subscriber = Subscriber.create(subscriberSettings);

// Start the subscriber. Upon successful starting, its state will become RUNNING.
subscriber.startAsync().awaitRunning();

System.out.println("Listening to messages on " + subscriptionPath.toString() + "...");

try {
  System.out.println(subscriber.state());
  // Wait 90 seconds for the subscriber to reach TERMINATED state. If it encounters
  // unrecoverable errors before then, its state will change to FAILED and an
  // IllegalStateException will be thrown.
  subscriber.awaitTerminated(90, TimeUnit.SECONDS);
} catch (TimeoutException t) {
  // Shut down the subscriber. This will change the state of the subscriber to TERMINATED.
  subscriber.stopAsync().awaitTerminated();
  System.out.println("Subscriber is shut down: " + subscriber.state());
}