Introduction

Welcome back! In this lesson, we will explore how to implement the fan-out messaging pattern using Google Cloud messaging services. The fan-out pattern allows a single publisher to send messages to multiple independent subscribers, enabling efficient and scalable communication between different parts of your application. By leveraging topics and multiple subscriptions, you can ensure that each subscriber receives a copy of every message published.

Initial setup

Consider a scenario with three services: Service A, Service B, and Service C. Service A needs to send updates to both Service B and Service C. Instead of sending messages directly to each service, we can create a topic that acts as a central channel. Both Service B and Service C can then subscribe to this topic, ensuring that they each receive all messages published by Service A.

Publishing to a Topic

Let's see how Service A can publish messages to a topic. We'll also include additional message attributes to provide more context to subscribers.

from google.cloud import pubsub_v1

# Initialize the publisher client
publisher = pubsub_v1.PublisherClient()

# Define the topic path
project_id = "your-gcp-project-id"
topic_id = "service-a-updates"
topic_path = publisher.topic_path(project_id, topic_id)

# Create the topic (if it doesn't exist)
try:
    publisher.create_topic(request={"name": topic_path})
    print(f"Topic created: {topic_path}")
except Exception:
    print(f"Topic already exists: {topic_path}")

# Publish a basic message
future = publisher.publish(topic_path, b"Updates for Services B and C")
print(f"Published message ID: {future.result()}")

# Publish a message with additional attributes
future_advanced = publisher.publish(
    topic_path,
    b"News update",
    priority="high",
    department="IT"
)
print(f"Published advanced message ID: {future_advanced.result()}")

Output:

Topic created: projects/your-gcp-project-id/topics/service-a-updates
Published message ID: 8234567890123456789
Published advanced message ID: 8234567890123456790

In these examples, a message is published to the service-a-updates topic. The second message includes attributes such as priority and department, which can be used by subscribers to filter or process messages accordingly.

Creating Multiple Subscriptions for Fan-Out

To ensure that both Service B and Service C receive all messages from Service A, we create separate subscriptions to the same topic. Each subscription acts as an independent channel, allowing each service to process messages at its own pace.

from google.cloud import pubsub_v1

# Initialize the subscriber client
subscriber = pubsub_v1.SubscriberClient()

# Define subscription paths
subscription_b_id = "service-b-sub"
subscription_c_id = "service-c-sub"
subscription_b_path = subscriber.subscription_path(project_id, subscription_b_id)
subscription_c_path = subscriber.subscription_path(project_id, subscription_c_id)

# Create subscriptions (if they don't exist)
for subscription_path in [subscription_b_path, subscription_c_path]:
    try:
        subscriber.create_subscription(
            request={"name": subscription_path, "topic": topic_path}
        )
        print(f"Subscription created: {subscription_path}")
    except Exception:
        print(f"Subscription already exists: {subscription_path}")

print("Both subscriptions are ready to receive messages!")

# Example: Pull messages for Service B
response = subscriber.pull(
    request={
        "subscription": subscription_b_path,
        "max_messages": 10,
    }
)

print(f"Pulled {len(response.received_messages)} messages for Service B:")

for received_message in response.received_messages:
    message = received_message.message
    print(f"Message ID: {message.message_id}")
    print(f"Data: {message.data.decode('utf-8')}")
    print(f"Attributes: {dict(message.attributes)}")
    print(f"Publish time: {message.publish_time}")
    print("---")
    
    # Acknowledge the message
    subscriber.acknowledge(
        request={
            "subscription": subscription_b_path,
            "ack_ids": [received_message.ack_id],
        }
    )

print("All messages acknowledged for Service B")

# Example: Pull messages for Service C to demonstrate fan-out
response_c = subscriber.pull(
    request={
        "subscription": subscription_c_path,
        "max_messages": 10,
    }
)

print(f"\nPulled {len(response_c.received_messages)} messages for Service C:")

for received_message in response_c.received_messages:
    message = received_message.message
    print(f"Message ID: {message.message_id}")
    print(f"Data: {message.data.decode('utf-8')}")
    print(f"Attributes: {dict(message.attributes)}")
    print("---")
    
    # Acknowledge the message
    subscriber.acknowledge(
        request={
            "subscription": subscription_c_path,
            "ack_ids": [received_message.ack_id],
        }
    )

print("All messages acknowledged for Service C")

Output:

Subscription created: projects/your-gcp-project-id/subscriptions/service-b-sub
Subscription created: projects/your-gcp-project-id/subscriptions/service-c-sub
Both subscriptions are ready to receive messages!
Pulled 2 messages for Service B:
Message ID: 8234567890123456789
Data: Updates for Services B and C
Attributes: {}
Publish time: 2024-01-15 10:30:45.123456+00:00
---
Message ID: 8234567890123456790
Data: News update
Attributes: {'priority': 'high', 'department': 'IT'}
Publish time: 2024-01-15 10:30:46.789012+00:00
---
All messages acknowledged for Service B

Pulled 2 messages for Service C:
Message ID: 8234567890123456789
Data: Updates for Services B and C
Attributes: {}
Publish time: 2024-01-15 10:30:45.123456+00:00
---
Message ID: 8234567890123456790
Data: News update
Attributes: {'priority': 'high', 'department': 'IT'}
Publish time: 2024-01-15 10:30:46.789012+00:00
---
All messages acknowledged for Service C

By creating multiple subscriptions to the same topic, each service receives its own copy of every message. This approach demonstrates the fan-out pattern, ensuring reliable and independent message delivery to all subscribers. Notice how both Service B and Service C received the same messages with identical message IDs, confirming that the fan-out pattern is working correctly.

Summary

In this lesson, we explored how to implement the fan-out messaging pattern using Google Cloud messaging services. By creating a topic and multiple subscriptions, a single publisher can efficiently distribute messages to multiple subscribers. This pattern enables scalable and reliable communication between different components of your application, allowing each subscriber to process messages independently. Experiment with creating additional topics and subscriptions to further customize your messaging architecture.

Previous
Sign up
Join the 1M+ learners on CodeSignal
Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal