Cloud Pubsub Laravel Package

Getting Started

Minimal Setup

1. Installation:

   composer require google/cloud-pubsub
   

   For gRPC (recommended for streaming):

   composer require google/cloud-pubsub-grpc
   

2. First Use Case: Publish a message to a topic:

   use Google\Cloud\PubSub\V1\PublisherClient;
   use Google\Cloud\PubSub\V1\TopicName;
   
   $publisher = new PublisherClient();
   $topicName = new TopicName('your-project-id', 'your-topic-name');
   
   $future = $publisher->publish($topicName, ['data' => 'Hello, Pub/Sub!']);
   $messageId = $future->wait(); // Blocking call
   

3. Key Files:

   • API Reference
   • Authentication Guide
   • gRPC Setup (for streaming)

Implementation Patterns

Core Workflows

1. Producer Pattern (Publish Messages)

• Batching: Use PublisherClient with publish() for async batching (default: 100 messages or 1MB).

  $batchSettings = $publisher->batchSettings();
  $batchSettings->setMaxMessages(500)->setMaxBytes(5 * 1024 * 1024); // 5MB
  $publisher->setBatchSettings($batchSettings);
  

• Ordered Publishing: Set orderingKey in the message for FIFO delivery.

  $future = $publisher->publish($topicName, ['data' => 'data', 'orderingKey' => 'user-123']);
  

2. Consumer Pattern (Subscribe & Pull)

• Push Subscriptions: Configure HTTP endpoints (e.g., Cloud Functions) via:

  $subscription = $subscriber->subscription('projects/p/subscriptions/s');
  $subscription->create([
      'pushConfig' => ['pushEndpoint' => 'https://your-endpoint.com/pubsub']
  ]);
  

• Pull Subscriptions: Stream messages with StreamingPullClient (gRPC required):

  $streamingPullClient = new StreamingPullClient();
  $stream = $streamingPullClient->streamingPull($subscriptionName);
  foreach ($stream as $response) {
      foreach ($response->getReceivedMessages() as $message) {
          echo $message->getMessage()->getData(); // Process message
          $streamingPullClient->acknowledge($subscriptionName, [$message->getAckId()]);
      }
  }
  

• Ack/Nack Deadlines: Extend deadlines dynamically:

  $subscriber->modifyAckDeadline($subscriptionName, $message->getAckId(), 300); // 5min
  

3. Schema & Transformations

• Schema Validation: Attach a schema to a topic:

  $topic->create(['schemaSettings' => ['schema' => 'projects/p/topics/t/schemas/s']]);
  

• Message Transforms: Apply transformations (e.g., AI inference) via:

  $transform = new \Google\Cloud\PubSub\V1\MessageTransform();
  $transform->setDisabled(false);
  $transform->setType('AI_INFERENCE');
  $topic->create(['messageTransforms' => [$transform]]);
  

4. Error Handling & Retries

• Exponential Backoff: Use RetrySettings for pull operations:

  $retrySettings = $subscriber->retrySettings();
  $retrySettings->setMaxAttempts(5);
  $subscriber->setRetrySettings($retrySettings);
  

• Dead Letter Topics: Configure via subscription:

  $subscription->create(['deadLetterPolicy' => ['deadLetterTopic' => 'projects/p/topics/dlq']]);
  

5. Integration with Laravel

• Service Provider:

  use Google\Cloud\PubSub\V1\PublisherClient;
  use Illuminate\Support\ServiceProvider;
  
  class PubSubServiceProvider extends ServiceProvider {
      public function register() {
          $this->app->singleton(PublisherClient::class, function ($app) {
              return new PublisherClient();
          });
      }
  }
  

• Queue Worker: Extend Laravel’s queue system:

  class PubSubQueueWorker extends ShouldQueue {
      protected $topicName;
  
      public function __construct(string $topicName) {
          $this->topicName = $topicName;
      }
  
      public function handle() {
          $publisher = app(PublisherClient::class);
          $publisher->publish($this->topicName, ['data' => $this->payload]);
      }
  }
  

Gotchas and Tips

Pitfalls

1. Authentication:

   • Avoid Hardcoding Credentials: Use environment variables or Google’s Application Default Credentials.
   • Deprecated Options: keyFile and keyFilePath are deprecated; use credentials instead:

     $client = new PublisherClient(['credentials' => $serviceAccount]);
     

2. gRPC vs REST:

   • Streaming Requires gRPC: For StreamingPullClient, ensure gRPC is installed. Fallback to REST for simple pull:

     $subscriber = new SubscriberClient();
     $response = $subscriber->pull($subscriptionName, ['returnImmediately' => false]);
     

3. Message Size Limits:

   • 10MB Hard Limit: Messages >10MB will fail. Use Cloud Storage subscriptions for large payloads.
   • Batching Limits: Default batch size is 100 messages/1MB. Adjust via batchSettings().

4. Idempotency:

   • Ack/Nack Race Conditions: Always use ackId for acknowledgments. Never rely on message content.
   • Duplicate Messages: Pub/Sub may redeliver messages. Design consumers to be idempotent.

5. Schema Mismatches:

   • Ingestion Failures: Schema violations trigger IngestionFailureEvent. Monitor with:

     $subscription->getIamPolicy(); // Check for monitoring permissions
     

Debugging

1. Enable Logging:

   $client = new PublisherClient([
       'logger' => new \Monolog\Logger('pubsub', [
           new \Monolog\Handler\StreamHandler('php://stderr', \Monolog\Logger::DEBUG)
       ])
   ]);
   

   • Logs gRPC/REST traffic, retries, and errors.

2. Common Errors:

   • INVALID_ARGUMENT: Validate topic/subscription names (format: projects/{project}/topics/{topic}).
   • PERMISSION_DENIED: Ensure IAM roles (roles/pubsub.publisher, roles/pubsub.subscriber).
   • DEADLINE_EXCEEDED: Increase timeout for large pulls:

     $subscriber->setGrpcOptions(['deadline' => new \Google\ApiCore\Grpc\Deadline(60)]);
     

3. Testing:

   • Local Emulation: Use Pub/Sub Emulator for CI:

     docker run -p 8085:8085 gcr.io/google.com/cloudsdktool/cloud-pubsub-emulator
     

   • Mock Clients: Stub PublisherClient in tests:

     $mock = $this->createMock(PublisherClient::class);
     $mock->method('publish')->willReturn(new \Google\ApiCore\Future\Future());
     

Extension Points

1. Custom Message Handlers:

   • Extend Message class for domain-specific logic:

     class OrderMessage extends \Google\Cloud\PubSub\V1\PubsubMessage {
         public function getOrderId(): string {
             return json_decode($this->getData(), true)['order_id'];
         }
     }
     

2. Middleware for Consumers:

   • Wrap StreamingPullClient to add preprocessing:

     $stream = $streamingPullClient->streamingPull($subscriptionName);
     foreach ($stream as $response) {
         foreach ($response->getReceivedMessages() as $message) {
             $processed = $this->preprocess($message->getMessage());
             $this->handle($processed);
             $streamingPullClient->acknowledge($subscriptionName, [$message->getAckId()]);
         }
     }
     

3. Event Dispatching:

   • Integrate with Laravel Events:

     event(new PubSubMessageReceived($message