Cloud Pubsub Laravel Package

Getting Started

Minimal Setup

1. Installation:

   composer require google/cloud-pubsub
   

   Ensure your Laravel project uses PHP 8.1+ (recommended) for full compatibility.

2. Authentication: Configure credentials via environment variables (recommended) or Laravel's .env:

   GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json
   

   Or use the GOOGLE_CLOUD_PROJECT environment variable for ADC (Application Default Credentials).

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

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

Key Entry Points

• Publisher: PublisherClient for sending messages.
• Subscriber: SubscriberClient for pulling messages (with streaming support).
• Admin: TopicAdminClient/SubscriptionAdminClient for managing resources.
• Batch: BatchingPublisherClient for optimized bulk publishing.

Implementation Patterns

Core Workflows

1. Publishing Messages

• Synchronous:

  $future = $publisher->publish($topic, ['data' => json_encode($payload)]);
  $messageId = $future->wait(); // Blocks until published
  

• Asynchronous (Recommended for Laravel Queues):

  $future = $publisher->publish($topic, ['data' => $payload]);
  $future->then(function ($messageId) {
      Log::info("Published message ID: $messageId");
  });
  

• Batched Publishing (for high throughput):

  $batchSettings = new \Google\Cloud\PubSub\V1\BatchingSettings();
  $batchSettings->setElementCountLimit(100);
  $batchSettings->setDelayThreshold(\Google\Protobuf\Duration::fromSeconds(1));
  
  $batchPublisher = new \Google\Cloud\PubSub\V1\BatchingPublisherClient($batchSettings);
  $future = $batchPublisher->publish($topic, ['data' => $payload]);
  

2. Subscribing to Messages

• Pull Subscription (Manual ACK):

  $subscriber = new \Google\Cloud\PubSub\V1\SubscriberClient();
  $subscription = $subscriber->subscriptionName('project-id', 'sub-name');
  
  $response = $subscriber->pull($subscription, ['maxMessages' => 10]);
  foreach ($response->getReceivedMessages() as $receivedMessage) {
      $message = $receivedMessage->getMessage();
      // Process message
      $subscriber->acknowledge($subscription, [$receivedMessage]);
  }
  

• Streaming Pull (Long-Running):

  $stream = $subscriber->streamingPull($subscription);
  foreach ($stream as $batch) {
      foreach ($batch->getReceivedMessages() as $receivedMessage) {
          $message = $receivedMessage->getMessage();
          // Process message
          $subscriber->acknowledge($subscription, [$receivedMessage]);
      }
  }
  

  Tip: Use Laravel's queue:work process manager to handle streaming subscribers.

3. Resource Management

• Create/Update Topics:

  $topicAdmin = new \Google\Cloud\PubSub\V1\TopicAdminClient();
  $topic = $topicAdmin->createTopic('project-id', 'new-topic');
  

• Create Subscriptions:

  $subscriptionAdmin = new \Google\Cloud\PubSub\V1\SubscriptionAdminClient();
  $subscription = $subscriptionAdmin->createSubscription(
      'project-id',
      'sub-name',
      'projects/project-id/topics/topic-name'
  );
  

4. Message Transformations (Advanced)

• Configure transformations when creating a subscription:

  $transform = new \Google\Cloud\PubSub\V1\MessageTransform();
  $transform->setEnabled(true);
  $transform->setSchema('projects/project-id/schemas/schema-name');
  
  $subscription = $subscriptionAdmin->createSubscription(
      'project-id',
      'sub-name',
      'projects/project-id/topics/topic-name',
      ['messageTransform' => $transform]
  );
  

Laravel Integration Tips

1. Service Provider: Bind the client to Laravel's container in AppServiceProvider:

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

2. Queued Jobs: Use Laravel Queues to process Pub/Sub messages asynchronously:

   class ProcessPubSubMessage implements ShouldQueue
   {
       public function handle()
       {
           $subscriber = resolve(\Google\Cloud\PubSub\V1\SubscriberClient::class);
           $subscription = $subscriber->subscriptionName('project-id', 'sub-name');
   
           $response = $subscriber->pull($subscription, ['maxMessages' => 1]);
           if ($response->getReceivedMessages()) {
               $this->dispatch(new ProcessMessageJob($response->getReceivedMessages()[0]));
           }
       }
   }
   

3. Event Dispatching: Trigger Laravel events from Pub/Sub messages:

   $subscriber->pull($subscription, ['maxMessages' => 10])->then(function ($response) {
       foreach ($response->getReceivedMessages() as $receivedMessage) {
           event(new PubSubMessageReceived($receivedMessage->getMessage()));
           $subscriber->acknowledge($subscription, [$receivedMessage]);
       }
   });
   

4. Retry Logic: Implement exponential backoff for failed operations:

   use Google\ApiCore\ApiException;
   use Symfony\Component\Process\Exception\ProcessFailedException;
   
   try {
       $future = $publisher->publish($topic, ['data' => $payload]);
       $future->wait();
   } catch (ApiException $e) {
       if ($e->getStatus()->getCode() === 5) { // Retry on unavailable
           throw new ProcessFailedException($e);
       }
       throw $e;
   }
   

Gotchas and Tips

Common Pitfalls

1. Authentication Issues:

   • Error: Google\Auth\Exception\GoogleAuthException: Could not load credentials.
   • Fix: Ensure GOOGLE_APPLICATION_CREDENTIALS is set or use ADC. Verify the service account has roles/pubsub.publisher/roles/pubsub.subscriber.
   • Laravel-Specific: Use Laravel's env() helper to load credentials dynamically:

     putenv('GOOGLE_APPLICATION_CREDENTIALS=' . env('GOOGLE_CREDENTIALS_PATH'));
     

2. Message Size Limits:

   • Pub/Sub enforces a 10MB message limit. For larger payloads, use Cloud Storage subscriptions or compress data.
   • Workaround: Store large data in GCS and publish a reference (e.g., GCS URI) as the message payload.

3. Streaming Pull Timeouts:

   • Default timeout is 60 seconds. Increase with:

     $stream = $subscriber->streamingPull($subscription, [
         'request' => new \Google\Cloud\PubSub\V1\StreamingPullRequest([
             'maxMessages' => 100,
             'returnImmediately' => false,
         ])
     ]);
     

   • For Laravel, use a separate process (e.g., queue:listen --daemon) to avoid HTTP timeout issues.

4. Message Ordering:

   • Pub/Sub does not guarantee order across different messages. Use ordering keys for sequential processing:

     $future = $publisher->publish($topic, [
         'data' => $payload,
         'orderingKey' => 'user-' . $userId
     ]);
     

5. Deprecated Options:

   • Avoid keyFile/keyFilePath in client options (deprecated in v2.15.0). Use environment variables or ADC instead.

Debugging Tips

1. Enable Debug Logging:

   $client = new \Google\Cloud\PubSub\V1\PublisherClient([
       'logger' => new \Monolog\Logger('pubsub', [
           new \Monolog\Handler\StreamHandler(storage_path('logs/pubsub.log'), \Monolog\Logger::DEBUG),
       ]),
   ]);
   

2. Inspect API Responses:

   • Use serializeToJsonString() to debug complex responses: