C3 AI Documentation Home

File on Arrival Mechanism for GCS

When a new file lands in the GCS bucket, the bucket can send a notification to a Pub/Sub topic. Starting onArrival mode on a FileSourceCollection adds a notification configuration to the bucket for new objects (the OBJECT_FINALIZE event).

Subscribers to the Pub/Sub topic receive messages with details about the uploaded file and process them.

To set up file on arrival, you prepare Pub/Sub, start onArrival mode, and register the subscriber. When you no longer need the setup, you tear it down.

Prepare Pub/Sub service

Prepare Pub/Sub in your Google Cloud Platform (GCP) project before you start onArrival mode:

  1. Make sure that the Pub/Sub service is set up in your GCP project.

  2. Create the Pub/Sub topic that receives the bucket notifications, and a subscription to that topic. You can create them in the Google Cloud console or from your application, as Manage GCP from User Applications describes.

  3. Point the platform at your subscription and topic. The platform reads two configurations of CloudMessageDispatcherConfig, each with two fields:

    ConfigurationcloudMessageBrokerName (default)cloudMessageBrokerType
    GcpBucketEvent${cluster}-${env}-${app}-GcpBucketEvent, your subscriptionGcpSubscription
    GcpBucketRegistration${cluster}-${env}-${app}-GcpBucketRegistration, your topicGcpTopic

    If your subscription and topic use these default names, you need no override. Otherwise, add config/cloudMessageDispatcher/GcpBucketEvent.json and config/cloudMessageDispatcher/GcpBucketRegistration.json to your application package with your names, for example:

    JSON
    {
  "cloudMessageBrokerName": "<your subscription name>",
  "cloudMessageBrokerType": "GcpSubscription"
}

    Or set the value at runtime in the Console:

    JavaScript
    CloudMessageDispatcherConfig.forName('GcpBucketEvent')
  .setConfigValue('cloudMessageBrokerName', '<your subscription name>', ConfigOverride.APP);

  4. Make sure that the GCP credentials you use have the following roles:

    • Storage Admin (roles/storage.admin): grants permissions to manage the GCS bucket.
    • Pub/Sub Admin (roles/pubsub.admin): grants permissions to manage Pub/Sub topics and subscriptions.

Key considerations

Check these points when you set up the mechanism:

  • Event types: onArrival mode sets the bucket to send an event only when an object is created or overwritten (OBJECT_FINALIZE). Deleting a file sends no event.
  • Testing: After you set everything up, upload a file to the collection's inbox directory and confirm that your app receives a notification.

Handle incoming files

Set up the C3 AI infrastructure to handle incoming files in your GCS bucket. You define a FileSourceSystem and a FileSourceCollection in it. The collection points to a directory in the bucket, its inbox, and your application processes the files that land there.

  1. Connect your application to the GCS bucket: enable the GCS file system, create a mount for the bucket, and set the bucket credentials, as Google Cloud Storage Connector describes.
  2. Create a FileSourceSystem whose root URL points to the bucket, and a FileSourceCollection whose inbox is the directory that receives the files. Steps 1 and 4 of Ingest Files from Amazon S3 Programmatically show both for an Amazon S3 bucket. For GCS, use the gcs:// URL of your bucket instead.

If the FileSourceSystem and FileSourceCollection already exist, do not create new ones. To enable onArrival mode or change its directory, update the existing configuration directly. Use FileSourceCollection.forName() to access your existing collection.

Start onArrival mode

In onArrival mode, the collection processes each file as it arrives. Start onArrival mode on your FileSourceCollection (fsc) by running fsc.startOnArrivalMode(). This method registers the collection's inbox URL (fsc.inboxUrl()) for file system notifications, so the bucket sends an event for each new file in that directory.

Once registered, each event goes to the Pub/Sub topic defined in your GcpBucketRegistration.json file.

Your application can then process new files as they arrive.

Register CloudInboundMessage

Register a CloudInboundMessage so that messages flow from your Google Cloud Pub/Sub setup to your application.

Overview of Pub/Sub data flow

The data flow works as follows:

  1. The data flow in Pub/Sub includes the following components:
    • Publisher: The entity that sends messages, in this case your GCS bucket.
    • Topic: The channel that carries the messages.
    • Subscription: The mechanism that lets subscribers receive messages from a topic.
    • Subscriber: The entity that receives messages from the subscription.
  2. When you call fsc.startOnArrivalMode(), the GCS bucket becomes a publisher: each new file in the inbox directory publishes a message to the topic.
  3. After the topic receives messages, GCP delivers them to the subscription.

Register the subscriber

To specify the subscriber and enable it to pull messages from the subscription, run the following command:

JavaScript
CloudInboundMessage.downcast(GcpBucketEvent.myType()).register();

This line registers the message Type that your application handles, in this case GcpBucketEvent.

After registration, the CloudInvalidationQueue pulls messages from the GCP subscription periodically, so your application processes file upload events as they occur.

Teardown

Clean up the setup by stopping onArrival mode and unregistering the message Type.

Stop onArrival mode

Run fsc.stopOnArrivalMode(). This method deregisters the URL you provided in the FileSourceCollection, so the GCS bucket stops publishing file change events to the Pub/Sub topic.

Clear the CloudInvalidationQueue

Next, unregister GcpBucketEvent by running the following command:

JavaScript
CloudInboundMessage.downcast(GcpBucketEvent.myType()).unregister();

This call removes the registration for GcpBucketEvent. The call also removes the CloudInvalidationQueue entries for the subscription when no other registered message Type uses it, so your application stops pulling these messages.

See also

Was this page helpful?