Implementing the Fan Out Pattern

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. Pythonfrom 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()}")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: textTopic created: projects/your-gcp-project-id/topics/service-a-updates Published message ID: 8234567890123456789 Published advanced message ID: 8234567890123456790Topic 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.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Python

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()}")

text

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

Python

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")

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")

text

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

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

Python

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")

text

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

Introduction

Initial setup

Publishing to a Topic

Join the 1M+ learners on CodeSignal

Creating Multiple Subscriptions for Fan-Out

Summary