Reading Data

This tutorial will show the basics of reading data from a MongoDB server. It assumes that you have read the Connecting to a MongoDB Server tutorial, and that you have a MongoDB server that contains data you wish to read.

Declaring Application State Type

This program will require that certain objects outlive the sub-operations, so we will need to persist them outside the event loop and pass them to our continuation. We declare a simple aggregate type to hold these objects:

Application State Struct

// Application state
typedef struct {
    // The client
    amongoc_client* client;
    // The collection we are reading from
    amongoc_collection* collection;
    // The name of the database we read from
    const char* db_name;
    // The name of the collection within the database
    const char* coll_name;
    int         batch_num;
} app_state;

Parsing Command Arguments

Our program will take three parameters: A MongoDB server URI, a database name, and a collection name. We can parse those as the start of main:

Argument Handling

int main(int argc, char** argv) {
    // Program parameters
    if (argc != 4) {
        fprintf(stderr, "Usage: %s <uri> <database> <collection>\n", argv[0]);
        return 2;
    }
    // The server URI
    const char* const uri = argv[1];

    // Initialize our state
    app_state state = {
        .db_name   = argv[2],
        .coll_name = argv[3],
    };

We store the database name and the collection name in the shared state struct so that we can access it in subsequent operations.

Starting with a Connection

We next do the basics of setting up an event loop and creating a connection emitter:

Setting up an event loop

    amongoc_loop loop;
    amongoc_if_error (amongoc_default_loop_init(&loop), msg) {
        fprintf(stderr, "Error initializing the event loop: %s\n", msg);
        return 1;
    }

    // Connect
    amongoc_emitter em = amongoc_client_new(&loop, uri);

Create the First Continuation

We have the pending connection object and the application state, so now we can set the first continuation:

The first continuation

    // Set the continuation
    em = amongoc_let(em, amongoc_async_forward_errors, amongoc_box_pointer(&state), on_connect);

The arguments are as follows:

1. Passing the emitter we got from amongoc_client_new(), and replacing it with the new operation emitter returned by amongoc_let().

2. amongoc_async_forward_errors is a behavior control flag for amongoc_let() that tells the operation to skip our continuation if the input operation generates an error.

3. We pass the address of the state object by wrapping it with amongoc_box_pointer(), passing it as the userdata parameter of amongoc_let().

4. on_connect is the name of the function that will handle the continuation.

Define the Continuation

Now we can look at the definition of on_connect:

Continuation signature

// Continuation after connection completes
static amongoc_emitter on_connect(amongoc_box state_, amongoc_status status, amongoc_box client) {
    // We don't use the status
    (void)status;

The state_ parameter is the userdata box that was given to amongoc_let() in the previous section. The client parameter is a box that contains the amongoc_client pointer from the connection operation. The status parameter here is the status of the connection operation, but we don’t care about this here since we used amongoc_async_forward_errors to ask amongoc_let() to skip this function if the status would otherwise indicate an error (i.e. on_connect will only be called if the connection actually succeeds).

Update the Application State

on_connect is passed the pointer to the application state as well as a box containing a pointer to the amongoc_client. We can extract the box value and store the client pointer within our application state struct:

Update the Application State

    // Store the client object
    app_state* state = amongoc_box_cast(app_state*, state_);
    amongoc_box_take(state->client, client);

We store it in the application state object so that we can later delete the client handle when the program completes.

Create a Collection Handle

Now that we have a valid client, we can create a handle to a collection on the server:

Set up the Collection

    // Create a new collection handle
    state->collection = amongoc_collection_new(state->client, state->db_name, state->coll_name);

amongoc_collection_new() does not actually communicate with the server: It only creates a client-side handle that can be used to perform operations related to that collection.

We store the returned collection handle in the state struct so that we can later delete it.

Initiate a Read Operation

There are several different reading operations and reading parameters. For this program, we’ll simply do a “find all” operation:

Start a “Find All” Operation

    // Initiate a read operation.
    amongoc_emitter em = amongoc_find(state->collection, bson_view_null, NULL);
    return amongoc_let(em, amongoc_async_forward_errors, state_, on_find);

The amongoc_find() function will return a new amongoc_emitter that represents the pending read from a collection. The first parameter state->collection is a pointer to a collection handle. The second parameter bson_view_null is a filter for the find operation. In this case, we are passing no filter, which is equivalent to an empty filter, telling the database that we want to find all documents in the collection. The third parameter is a set of common named parameters for find operations. In this case, we are passing NULL, because we don’t care to specify any more parameters.

We attach a second continuation using amongoc_let() to a function on_find.

Read from The Cursor

When the amongoc_find() emitter resolves, it resolves with an amongoc_cursor object. We’ll extract that in on_find:

Handle Data

// Continuation when we get data
static amongoc_emitter on_find(amongoc_box state_, amongoc_status status, amongoc_box cursor_) {
    // We don't use the status
    (void)status;
    app_state* state = amongoc_box_cast(app_state*, state_);
    // Extract the cursor
    amongoc_cursor cursor;
    amongoc_box_take(cursor, cursor_);

We can print out the cursor’s batch of data:

Print the Records

    // Print the data
    fprintf(stderr,
            "Got results from database '%s' collection '%s', batch %d: ",
            state->db_name,
            state->coll_name,
            state->batch_num);
    bson_write_repr(stderr, cursor.records);
    fprintf(stderr, "\n");
    state->batch_num++;

Read Some More

A single read may not return the full set of data, so we may need to initiate a subsequent read with amongoc_cursor_next():

Check for more

    // Do we have more data?
    if (cursor.cursor_id) {
        // More to find
        amongoc_emitter em = amongoc_cursor_next(cursor);
        return amongoc_let(em, amongoc_async_forward_errors, state_, on_find);
    } else {
        // Done
        amongoc_cursor_delete(cursor);
        return amongoc_just();
    }
}

By passing on_find to amongoc_let() again, we create a loop that continually calls on_find until the cursor ID is zero (indicating a finished read).

If we have no more data, we destroy the cursor object and return a null result immediately with amongoc_just().

Tie, Start, and Run the Operation

Going back to main, we can now tie the operation, start it, and run the event loop:

Run the Program

    // Run the program and collect the result
    amongoc_status    status;
    amongoc_operation op = amongoc_tie(em, &status);
    amongoc_start(&op);
    amongoc_default_loop_run(&loop);
    amongoc_operation_delete(op);
    amongoc_collection_delete(state.collection);
    amongoc_client_delete(state.client);
    amongoc_default_loop_destroy(&loop);

amongoc_tie() tells the emitter to store its final status and result value in the given destinations, and returns an operation that can be initiated with amongoc_start(). We then give control to the event loop with amongoc_default_loop_run(). After this returns, the operation object is finished, so we delete it with amongoc_operation_delete().

We are also finished with the collection handle and the client, so we delete those here as well. Those struct members will have been filled in by on_connect.

Check for Errors

If any sub-operation failed, the error status will be propagated to the final status, so we check that before exiting:

Check for Errors

    // Check for errors
    amongoc_if_error (status, msg) {
        fprintf(stderr, "Error: %s\n", msg);
        return 1;
    }

The Full Program

Here is the full sample program, in its entirety:

read.example.c

#include <amongoc/amongoc.h>

// Application state
typedef struct {
    // The client
    amongoc_client* client;
    // The collection we are reading from
    amongoc_collection* collection;
    // The name of the database we read from
    const char* db_name;
    // The name of the collection within the database
    const char* coll_name;
    int         batch_num;
} app_state;

static amongoc_emitter on_connect(amongoc_box state, amongoc_status status, amongoc_box client);

int main(int argc, char** argv) {
    // Program parameters
    if (argc != 4) {
        fprintf(stderr, "Usage: %s <uri> <database> <collection>\n", argv[0]);
        return 2;
    }
    // The server URI
    const char* const uri = argv[1];

    // Initialize our state
    app_state state = {
        .db_name   = argv[2],
        .coll_name = argv[3],
    };

    amongoc_loop loop;
    amongoc_if_error (amongoc_default_loop_init(&loop), msg) {
        fprintf(stderr, "Error initializing the event loop: %s\n", msg);
        return 1;
    }

    // Connect
    amongoc_emitter em = amongoc_client_new(&loop, uri);

    // Set the continuation
    em = amongoc_let(em, amongoc_async_forward_errors, amongoc_box_pointer(&state), on_connect);

    // Run the program and collect the result
    amongoc_status    status;
    amongoc_operation op = amongoc_tie(em, &status);
    amongoc_start(&op);
    amongoc_default_loop_run(&loop);
    amongoc_operation_delete(op);
    amongoc_collection_delete(state.collection);
    amongoc_client_delete(state.client);
    amongoc_default_loop_destroy(&loop);

    // Check for errors
    amongoc_if_error (status, msg) {
        fprintf(stderr, "Error: %s\n", msg);
        return 1;
    }
}

static amongoc_emitter on_find(amongoc_box state_, amongoc_status status, amongoc_box cursor_);

// Continuation after connection completes
static amongoc_emitter on_connect(amongoc_box state_, amongoc_status status, amongoc_box client) {
    // We don't use the status
    (void)status;
    // Store the client object
    app_state* state = amongoc_box_cast(app_state*, state_);
    amongoc_box_take(state->client, client);
    // Create a new collection handle
    state->collection = amongoc_collection_new(state->client, state->db_name, state->coll_name);
    // Initiate a read operation.
    amongoc_emitter em = amongoc_find(state->collection, bson_view_null, NULL);
    return amongoc_let(em, amongoc_async_forward_errors, state_, on_find);
}

// Continuation when we get data
static amongoc_emitter on_find(amongoc_box state_, amongoc_status status, amongoc_box cursor_) {
    // We don't use the status
    (void)status;
    app_state* state = amongoc_box_cast(app_state*, state_);
    // Extract the cursor
    amongoc_cursor cursor;
    amongoc_box_take(cursor, cursor_);

    // Print the data
    fprintf(stderr,
            "Got results from database '%s' collection '%s', batch %d: ",
            state->db_name,
            state->coll_name,
            state->batch_num);
    bson_write_repr(stderr, cursor.records);
    fprintf(stderr, "\n");
    state->batch_num++;

    // Do we have more data?
    if (cursor.cursor_id) {
        // More to find
        amongoc_emitter em = amongoc_cursor_next(cursor);
        return amongoc_let(em, amongoc_async_forward_errors, state_, on_find);
    } else {
        // Done
        amongoc_cursor_delete(cursor);
        return amongoc_just();
    }
}
// end:on_find