MongoDB Time Series With C++

1
git clone https://github.com/Microsoft/vcpkg.git
2
cd vcpkg
3
./bootstrap-vcpkg.sh
4
./vcpkg integrate install

1
./vcpkg install cpr:x64-windows

1
./vcpkg install cpr

1
# this is heuristically generated, and may not be correct
2
find_package(cpr CONFIG REQUIRED)
3
target_link_libraries(main PRIVATE cpr::cpr)

1
#pragma once
2
3
4
#include <mongocxx/client.hpp>
5
#include <bsoncxx/builder/list.hpp>
6
#include <bsoncxx/builder/stream/document.hpp>
7
#include <bsoncxx/json.hpp>
8
#include <mongocxx/uri.hpp>
9
#include <mongocxx/instance.hpp>
10
11
12
#include <iostream>
13
#include <cpr/cpr.h>
14
15
16
using namespace std;
17
18
19
std::string getEnvironmentVariable(std::string environmentVarKey)
20
{
21
	char* pBuffer = nullptr;
22
	size_t size = 0;
23
	auto key = environmentVarKey.c_str();
24
	
25
	// Use the secure version of getenv, ie. _dupenv_s to fetch environment variable.  
26
	if (_dupenv_s(&pBuffer, &size, key) == 0 && pBuffer != nullptr)
27
	{
28
		std::string environmentVarValue(pBuffer);
29
		free(pBuffer);
30
		return environmentVarValue;
31
	}
32
	else
33
	{
34
		return "";
35
	}
36
}
37
38
39
int getAQI(std::string city, std::string apiToken)
40
{
41
	// Call the API to get the air quality index.
42
	std::string aqiUrl = "https://api.waqi.info/feed/" + city + "/?token=" + apiToken;
43
	auto aqicnResponse = cpr::Get(cpr::Url{ aqiUrl });
44
45
46
	// Get the AQI from the response
47
	if(aqicnResponse.text.empty())
48
	{
49
		cout << "Error: Response is empty." << endl;
50
		return -1;
51
	}
52
	bsoncxx::document::value aqicnResponseBson = bsoncxx::from_json(aqicnResponse.text);
53
	auto aqi = aqicnResponseBson.view()["data"]["aqi"].get_int32().value;
54
	return aqi;
55
}
56
57
58
void saveToCollection(mongocxx::collection& collection, int aqi)
59
{
60
	auto timeStamp = bsoncxx::types::b_date(std::chrono::system_clock::now());
61
62
63
	bsoncxx::builder::stream::document aqiDoc = bsoncxx::builder::stream::document{};
64
	aqiDoc << "timestamp" << timeStamp << "aqi" << aqi;
65
	collection.insert_one(aqiDoc.view());
66
67
68
	// Log to the console window.
69
	cout << " TimeStamp: " << timeStamp << " AQI: " << aqi << endl;
70
}

1
int main()
2
{
3
	// Get the required parameters from environment variable.
4
	auto mongoURIStr = getEnvironmentVariable("MONGODB_URI");
5
	auto apiToken = getEnvironmentVariable("AQICN_TOKEN");
6
	std::string city = "Delhi";
7
	static const mongocxx::uri mongoURI = mongocxx::uri{ mongoURIStr };
8
9
10
	if (mongoURI.to_string().empty() || apiToken.empty())
11
	{
12
		cout << "Invalid URI or API token. Please check the environment variables." << endl;
13
		return 0;
14
	}
15
16
17
	// Create an instance.
18
	mongocxx::instance inst{};
19
	mongocxx::options::client client_options;
20
	auto api = mongocxx::options::server_api{ mongocxx::options::server_api::version::k_version_1 };
21
	client_options.server_api_opts(api);
22
	mongocxx::client conn{ mongoURI, client_options };
23
24
25
	// Setup Database and Collection.
26
	const string dbName = "AQIMonitor";
27
	const string timeSeriesCollectionName = "AQIMonitorCollection";
28
29
30
	// Setup Time Series collection options.
31
	bsoncxx::builder::document timeSeriesCollectionOptions =
32
	{
33
	   "timeseries",
34
		{
35
		  "timeField", "timestamp",
36
		  "granularity", "minutes"
37
		}
38
	};
39
40
41
	auto aqiMonitorDB = conn[dbName];
42
	auto aqiMonitorCollection = aqiMonitorDB.has_collection(timeSeriesCollectionName)
43
		? aqiMonitorDB[timeSeriesCollectionName]
44
		: aqiMonitorDB.create_collection(timeSeriesCollectionName, timeSeriesCollectionOptions.view().get_document().value);
45
46
47
	// Fetch and update AQI every 30 minutes.
48
	while (true)
49
	{ 
50
		auto aqi = getAQI(city, apiToken);
51
		saveToCollection(aqiMonitorCollection, aqi);
52
		std::this_thread::sleep_for(std::chrono::minutes(30));
53
	}
54
55
56
	return 0;
57
}