Assignment 3 - Add pub/sub messaging

October 16, 2024 · View on GitHub

Assignment goals

To complete this assignment, you must reach the following goals:

The TrafficControlService sends SpeedingViolation messages using the Dapr pub/sub building block.
The FineCollectionService receives SpeedingViolation messages using the Dapr pub/sub building block.
RabbitMQ is used as pub/sub message broker that runs as part of the solution in a Docker container.

Don't worry if you have no experience with RabbitMQ. You will run it as a container in the background and don't need to interact with it directly in any way. The instructions will explain exactly how to do that.

This assignment targets number 2 in the end-state setup:

Step 1: Run RabbitMQ as message broker

In the example, you will use RabbitMQ as the message broker with the Dapr pub/sub building block. You're going to pull a standard Docker image containing RabbitMQ to your machine and start it as a container.

Open the terminal window in VS Code.

Start a RabbitMQ message broker by entering the following command:

docker run -d -p 5672:5672 -p 15672:15672 --name dtc-rabbitmq rabbitmq:3-management-alpine

This will pull the docker image rabbitmq:3-management-alpine from Docker Hub and start it. The name of the container will be dtc-rabbitmq. The server will be listening for connections on port 5672 (which is the default port for RabbitMQ).

If everything goes well, you should see some output like this:

❯ docker run -d -p 5672:5672 --name dtc-rabbitmq rabbitmq:3-management-alpine
Unable to find image 'rabbitmq:3-management-alpine' locally
3-management-alpine: Pulling from library/rabbitmq
a0d0a0d46f8b: Pull complete
31312314eeb3: Pull complete
926937e20d4d: Pull complete
f5676ddf0782: Pull complete
ff9526ce7ab4: Pull complete
6163319fe438: Pull complete
592def0a276e: Pull complete
59922d736a7b: Pull complete
76025ca84b3c: Pull complete
4965e42a5d3c: Pull complete
Digest: sha256:8885c08827289c61133d30be68658b67c6244e517931bb7f1b31752a9fcaec73
Status: Downloaded newer image for rabbitmq:3-management-alpine
85a98f00f1a87b856008fec85de98c8412eb099e3a7675b87945c777b131d876

If you see any errors, make sure you have access to the Internet and are able to download images from Docker Hub. See Docker Hub for more info.

The container will keep running in the background. If you want to stop it, enter the following command:

docker stop dtc-rabbitmq

You can then start the container later by entering the following command:

docker start dtc-rabbitmq

If you are done using the container, you can also remove it by entering the following command:

docker rm dtc-rabbitmq -f

Once you have removed it, you need to start it again with the docker run command shown at the beginning of this step.

For your convenience, the Infrastructure folder contains Powershell scripts for starting the infrastructural components you'll use throughout the workshop. You can use the Infrastructure/rabbitmq/start-rabbitmq.ps1 script to start the RabbitMQ container.

If you don't mind starting all the infrastructural containers at once (also for assignments to come), you can also use the Infrastructure/start-all.ps1 script.

Step 2: Configure the pub/sub component

Until now, you have been using the Dapr components that are installed by default when you install Dapr on your machine. These are a state management component and a pub/sub component. They both use the Redis server that is also installed by default. The components are installed in the folder %USERPROFILE%\.dapr\components on Windows and $HOME/.dapr/components on Linux or Mac.

Because you need to change the message broker from Redis to RabbitMQ, you will create a separate folder with the component configuration files and use this folder when starting the services using the Dapr CLI. You can specify which folder to use on the command-line with the --resources-path flag.

Create a new folder dapr/components.
Copy all files from the folder %USERPROFILE%\.dapr\components\ on Windows and $HOME/.dapr/components on Linux or Mac to the dapr/components folder.
Open the file dapr/components/pubsub.yaml in VS Code.
Inspect this file. As you can see, it specifies the type of the message broker to use (pubsub.redis) and specifies information on how to connect to the Redis server in the metadata section.

Change the content of this file to:

apiVersion: dapr.io/v1alpha1
kind: Component
metadata:
  name: pubsub
spec:
  type: pubsub.rabbitmq
  version: v1
  metadata:
  - name: host
    value: "amqp://localhost:5672"
  - name: durable
    value: "false"
  - name: deletedWhenUnused
    value: "false"
  - name: autoAck
    value: "false"
  - name: reconnectWait
    value: "0"
  - name: concurrency
    value: parallel
scopes:
  - trafficcontrolservice
  - finecollectionservice

As you can see, you specify a different type of pub/sub component (pubsub.rabbitmq) and you specify in the metadata how to connect to the RabbitMQ container you started in step 1 (running on localhost on port 5672). The other metadata can be ignored for now. In the scopes section, you specify that only the TrafficControlService and FineCollectionService should use the pub/sub building block.

Step 3: Send messages from the TrafficControlService

With the Dapr pub/sub building block, you use a topic to send and receive messages. The producer sends messages to the topic and one or more consumers subscribe to this topic to receive those messages. First you are going to prepare the TrafficControlService so it can send messages using Dapr pub/sub.

Open the file TrafficControlService/Controllers/TrafficController.cs in VS Code.

Near the end of the VehicleExit method, you find the code that sends a SpeedingViolation message to the collectfine endpoint of the FineCollectionService over HTTP:

// publish speedingviolation
var message = JsonContent.Create<SpeedingViolation>(speedingViolation);
await _httpClient.PostAsync("http://localhost:6001/collectfine", message);

The URL for publishing a message using the Dapr pub/sub API is: http://localhost:<daprPort>/v1.0/publish/<pubsub-name>/<topic>. You'll use this API to send a message to the speedingviolations topic using the pub/sub component named pubsub. The Dapr sidecar for the TrafficControlService runs on HTTP port 3600. Replace the URL in the HTTP call with a call to the Dapr pub/sub API:
```
await _httpClient.PostAsync("http://localhost:3600/v1.0/publish/pubsub/speedingviolations", message);
```

That's it. You now use Dapr pub/sub to publish a message to a message broker.

Step 4: Receive messages in the FineCollectionService (declaratively)

Now you are going to prepare the FineCollectionService so it can receive messages using Dapr pub/sub. Consuming messages can be done in two ways: declaratively (through configuration) or programmatic (from the code). First you'll use the declarative way. Later you'll also use the programmatic way and finally also using the Dapr SDK for .NET.

Add a new file in the dapr/components folder named subscription.yaml.
Open this file in VS Code.
You're going to define a subscription on a topic and link that to a web API operation on the FineCollectionService. Paste this snippet into the file:
```
apiVersion: dapr.io/v1alpha1
kind: Subscription
metadata:
  name: speedingviolations-subscription
spec:
  topic: speedingviolations
  route: /collectfine
  pubsubname: pubsub
scopes:
- finecollectionservice
```
The route field tells Dapr to forward all messages send to the speedingviolations topic to the /collectfine endpoint in the app. The scopes field restricts this subscription to only the service with app-id finecollectionservice only.

Now your FineCollectionService is ready to receive messages through Dapr pub/sub. But there is a catch! Dapr uses the CloudEvents message format for pub/sub. So when we send a message through pub/sub, the receiving application needs to understand this format and handle the message as a CloudEvent. Therefore we need to change the code slightly. For now, you will parse the incoming JSON by hand (instead of ASP.NET Core model binding doing that for you). You will change this later when you will use the Dapr SDK for .NET.

Open the file FineCollectionService/Controllers/CollectionController.cs in VS Code.
Remove the SpeedingViolation parameter from the CollectFine method and replace this with a cloudevent parameter of type System.Text.Json.JsonDocument that is decorated with the [FromBody] attribute:
```
public async Task<ActionResult> CollectFine([FromBody] System.Text.Json.JsonDocument cloudevent)
```
This enables you to get to the raw JSON in the request.

Add the following code at the top of the method to extract the SpeedingViolation data from the cloud event:

var data = cloudevent.RootElement.GetProperty("data");
var speedingViolation = new SpeedingViolation
{
    VehicleId = data.GetProperty("vehicleId").GetString()!,
    RoadId = data.GetProperty("roadId").GetString()!,
    Timestamp = data.GetProperty("timestamp").GetDateTime()!,
    ViolationInKmh = data.GetProperty("violationInKmh").GetInt32()
};

Open the terminal window in VS Code and make sure the current folder is FineCollectionService.
Check all your code-changes are correct by building the code. Execute the following command in the terminal window:
```
dotnet build
```
If you see any warnings or errors, review the previous steps to make sure the code is correct.

Step 5: Test the application

You're going to start all the services now. You specify the custom components folder you've created on the command-line using the --resources-path flag so Dapr will use these config files:

Make sure no services from previous tests are running (close the command-shell windows).
Open the terminal window in VS Code and make sure the current folder is VehicleRegistrationService.
Enter the following command to run the VehicleRegistrationService with a Dapr sidecar:
```
dapr run --app-id vehicleregistrationservice --app-port 6002 --dapr-http-port 3602 --dapr-grpc-port 60002 --resources-path ../dapr/components dotnet run
```
Notice that you specify the custom components folder you've created on the command-line using the --resources-path flag so Dapr will use RabbitMQ for pub/sub.
Open a new terminal window in VS Code and change the current folder to FineCollectionService.

Enter the following command to run the FineCollectionService with a Dapr sidecar:

dapr run --app-id finecollectionservice --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 --resources-path ../dapr/components dotnet run

Open a new terminal window in VS Code and change the current folder to TrafficControlService.

Enter the following command to run the TrafficControlService with a Dapr sidecar:

dapr run --app-id trafficcontrolservice --app-port 6000 --dapr-http-port 3600 --dapr-grpc-port 60000 --resources-path ../dapr/components dotnet run

Open a new terminal window in VS Code and change the current folder to Simulation.
Start the simulation:
```
dotnet run
```

You should see the same logs as before. Obviously, the behavior of the application is exactly the same as before. But if you look closely at the Dapr logs of the FineCollectionService, you should see something like this in there:

time="2021-02-27T16:46:02.5989612+01:00" level=info msg="app is subscribed to the following topics: [speedingviolations] through pubsub=pubsub" app_id=finecollectionservice instance=EDWINW01 scope=dapr.runtime type=log ver=1.0.0

So you can see that Dapr has registered a subscription for the FineCollectionService to the speedingviolations topic.

Step 6: Receive messages in the FineCollectionService (programmatic)

The other way of subscribing to pub/sub events is the programmatic way. Dapr will call your service on the well known endpoint /dapr/subscribe to retrieve the subscriptions for that service. You will implement this endpoint and return the subscription for the speedingviolations topic.

Stop the FineCollectionService by navigating to its terminal window and pressing Ctrl-C. You can keep the other services running for now.
Open the file FineCollectionService/Controllers/CollectionController.cs in VS Code.

Add a new operation named Subscribe to the controller that will listen to the route /dapr/subscribe:

[Route("/dapr/subscribe")]
[HttpGet()]
public object Subscribe()
{
    return new object[]
    {
        new
        {
            pubsubname = "pubsub",
            topic = "speedingviolations",
            route = "/collectfine"
        }
    };
}

Remove the file dapr/components/subscription.yaml. This file is not needed anymore because you implemented the /dapr/subscribe method.
Go back to the terminal window in VS Code and make sure the current folder is FineCollectionService.
Check all your code-changes are correct by building the code. Execute the following command in the terminal window:
```
dotnet build
```
If you see any warnings or errors, review the previous steps to make sure the code is correct.
Start the updated FineCollectionService:
```
dapr run --app-id finecollectionservice --app-port 6001 --dapr-http-port 3601 --dapr-grpc-port 60001 --resources-path ../dapr/components dotnet run
```
If you kept the other services running before you started with this step, you may observe that there a few speeding limitations waiting to be processed. The application will immediately start processing this backlog.
After you've looked at the log output and confirmed that everything works, you can stop all the services.

In this step, you will change the code slightly so it uses the Dapr SDK for .NET. First you'll change the TrafficControlService that sends messages.

Open the terminal window in VS Code and make sure the current folder is TrafficControlService.
Add a reference to the Dapr ASP.NET Core integration package:
```
dotnet add package Dapr.AspNetCore
```
Open the file TrafficControlService/GlobalUsings.cs in VS Code.
In this file, add a global using statement for the Dapr client:
```
global using Dapr.Client;
```
Open the file TrafficControlService/Controllers/TrafficController.cs in VS Code.
Add an argument named daprClient of type DaprClient that is decorated with the [FromServices] attribute to the VehicleExit method :
```
public async Task<ActionResult> VehicleExit(VehicleRegistered msg, [FromServices] DaprClient daprClient)
```
The [FromServices] attribute injects the DaprClient into the method using the ASP.NET Core dependency injection system.

Near the end of the method, you'll find the code that publishes the SpeedingViolation message using _httpClient:

// publish speedingviolation
var message = JsonContent.Create<SpeedingViolation>(speedingViolation);
await _httpClient.PostAsync("http://localhost:3600/v1.0/publish/pubsub/speedingviolations", message);

Replace this code with a call to the Dapr pub/sub building block using the DaprClient:

// publish speedingviolation
await daprClient.PublishEventAsync("pubsub", "speedingviolations", speedingViolation);

Open the file TrafficControlService/Program.cs.
The service now uses the DaprClient. Therefore, it needs to be registered with dependency injection. Insert the following code at line 3 into the file to register the DaprClient with dependency injection:
```
builder.Services.AddDaprClient(builder => builder
    .UseHttpEndpoint($"http://localhost:3600")
    .UseGrpcEndpoint($"http://localhost:60000"));
```
Open the terminal window in VS Code and make sure the current folder is TrafficControlService.
Check all your code-changes are correct by building the code. Execute the following command in the terminal window:
```
dotnet build
```
If you see any warnings or errors, review the previous steps to make sure the code is correct.

Now you will change the FineCollectionService that receives messages. The Dapr ASP.NET Core integration library offers an elegant way of linking an ASP.NET Core WebAPI method to a pub/sub topic. For every message sent to that topic, the WebAPI method is invoked and the payload of the message is delivered as request body. You don't have to poll for messages on the message broker.

Open the file FineCollectionService/Controllers/CollectionController.cs in VS Code.
Remove the Subscribe method from the controller.
Replace the cloudevent parameter of the CollectFine method with a parameter of type SpeedingViolation named speedingViolaton:
```
public async Task<ActionResult> CollectFine(SpeedingViolation speedingViolation)
```

Remove the code that parses the cloud event data at the beginning of the method:

var data = cloudevent.RootElement.GetProperty("data");
var speedingViolation = new SpeedingViolation
{
    VehicleId = data.GetProperty("vehicleId").GetString()!,
    RoadId = data.GetProperty("roadId").GetString()!,
    Timestamp = data.GetProperty("timestamp").GetDateTime()!,
    ViolationInKmh = data.GetProperty("violationInKmh").GetInt32()
};

Add a Topic attribute above the CollectFine method to link this method to a topic called speedingviolations:
```
[Topic("pubsub", "speedingviolations")]
```
The "pubsub" argument passed to this attribute refers to the name of the Dapr pub/sub component to use.
Open the file FineCollectionService/GlobalUsings.cs in VS Code.
In this file, add a global using statement for the Dapr library:
```
global using Dapr;
```

Now you need to make sure that Dapr knows this controller and also knows which pub/sub topics the controller subscribes to. To determine this, Dapr will call your service on a default endpoint to retrieve the subscriptions. To make sure your service handles this request and returns the correct information, you need to add some statements to the Startup class:

Open the file FineCollectionService/Program.cs in VS Code.
Add AddDapr to the AddControllers line in this file:
```
builder.Services.AddControllers().AddDapr();
```
The AddDapr method adds Dapr integration for ASP.NET MVC.
As you saw earlier, Dapr uses the cloud event message-format standard when sending messages over pub/sub. To make sure cloud events are automatically unwrapped, add the following line just after the call to app.MapControllers() in the file:
```
app.UseCloudEvents();
```
To register every controller that uses pub/sub as a subscriber, add a call to app.MapSubscribeHandler() just after the call to app.UseCloudEvent() in the file:
```
app.MapSubscribeHandler();
```
By adding this, the /dapr/subscribe endpoint that you implemented in step 6 is automatically implemented by Dapr. It will collect all the controller methods that are decorated with the Dapr Topic attribute and return the corresponding subscriptions.
Open the terminal window in VS Code and make sure the current folder is FineCollectionService.
Check all your code-changes are correct by building the code. Execute the following command in the terminal window:
```
dotnet build
```
If you see any warnings or errors, review the previous steps to make sure the code is correct.

Now you can test the application again. Execute the activities in step 5 again.

Next assignment

Make sure you stop all running processes and close all the terminal windows in VS Code before proceeding to the next assignment.

Go to assignment 4.