Consumers

So as of right now, we have the following setup:

A publisher publishes a message
The message arrives in an exchange
The message is routed to queue(s)
???
Profit

In all seriousness, nothing happens after the message arrives in the queue!

This is where consumers come in. Consumers are programs (like our "client" program) that connect to queues and pull the messages out of them.

Assignment

Let's configure our client consumers to process the "pause messages" and update their local game state.

In src/internal/pubsub, create a new function called subscribeJSON, here's my function signature:

export async function subscribeJSON<T>(
  conn: amqp.ChannelModel,
  exchange: string,
  queueName: string,
  key: string,
  queueType: SimpleQueueType, // an enum to represent "durable" or "transient"
  handler: (data: T) => void,
): Promise<void>;

In SubscribeJSON:
1. Call declareAndBind to make sure that the given queue exists and is bound to the exchange
2. Use the new ChannelModel to call the consume method.
  1. Use the queue returned by the declareAndBind call as the first parameter.
  2. Create a callback function that will be called each time a message is received. This callback function should:
    1. Have a message parameter of type amqp.ConsumeMessage.
    2. Use JSON.parse to parse the message content.
    3. Call the given handler function with the parsed message content.
    4. Acknowledge the message with channel.ack(message) to remove it from the queue.
Create a new function called handlerPause in src/client/index.ts. It accepts a game state and returns a new handler function that accepts a PlayingState. This will be the handler we pass into subscribeJSON that will be called each time a new message is consumed. Here's my signature:

function handlerPause(gs: GameState): (ps: PlayingState) => void;

In the handler returned from handlerPause:
1. Use game state handlePause function to pause the game for the client.
2. Print "> " at the end of the line so that the user can enter a new command.
Back in src/client/index.ts, in the main function, after creating the game state, call subscribeJSON with the following parameters:
1. The connection
2. The direct exchange (constant can be found in src/internal/routing/routing.ts)
3. A queue named pause.username where username is the username of the player
4. The routing key pause (constant can be found in src/internal/routing/routing.ts)
5. Transient queue type
6. The new handler we just created.
Test the New Code
1. Start an instance of the server and the client in separate terminals. Use washington as the username for the client.
2. Spawn a unit using the client: spawn europe infantry
3. Pause the game using the server. You should see the client detect (consume) the pause message
4. Try to move a unit, the client should not allow it because the game is paused
5. Resume the game using the server. You should see the client detect (consume) the resume message
6. Try to move a unit, the client should allow it

Run and submit the CLI tests with the server and client still running.

The Boot.dev CLI requires you to be signed in to submit your solution!

Copy/paste one of the following commands into your terminal:

Run

bootdev run e0b196c2-f9c4-4eca-b165-ced8bc8bc4fd

Submit

bootdev run e0b196c2-f9c4-4eca-b165-ced8bc8bc4fd -s

To run and submit the tests for this lesson, you must have an active Boot.dev membership

Using the Bootdev CLI

The Bootdev CLI is the only way to submit your solution for this type of lesson. We need to be able to run commands in your environment to verify your solution.

You can install it here. It's a Go program hosted on GitHub, so you'll need Go installed as well. Instructions are on the GitHub page.