Paths

Table of Contentst

-
services/tunnelbroker/src/Database/
-
tunnelbroker/
-
src/
-
Database/
-
MessageItem.cpp

[services] Tunnelbroker - Fix `GetN()`->`GetS()`.
ClosedPublic
Actions

Authored by • max on Apr 7 2022, 2:58 AM.

Details

Reviewers

• karol
tomek
varun

Commits

rCOMM46e5cc697cdb: [services] Tunnelbroker - Fix `GetN()`->`GetS()`.

Summary

This is a small fix error on std:stoull that throws an exception because it's expecting a string, but we return the number in GetN().
We've agreed before that we should use GetS() from AWS C++ SDK everywhere.
This diff changes GetN()->GetS().

Test Plan

Run yarn test-tunnelbroker-service and successfully build the app.

Diff Detail

Repository

rCOMM Comm

Branch

arcpatch-D3642

Lint

No Lint Coverage

Unit

No Test Coverage

Event Timeline

• max created this revision.Apr 7 2022, 2:58 AM

Herald added subscribers: • abosh, • benschac, • karol and 4 others. · View Herald TranscriptApr 7 2022, 2:58 AM

• max added reviewers: • karol, tomek, varun.Apr 7 2022, 2:58 AM

Harbormaster completed remote builds in B7956: Diff 11160.Apr 7 2022, 3:05 AM

• max requested review of this revision.Apr 7 2022, 3:05 AM

• max added a parent revision: D3643: [services] Tunnelbroker - Database item tests initialization.Apr 7 2022, 3:22 AM

• max removed a parent revision: D3643: [services] Tunnelbroker - Database item tests initialization.

• max added a child revision: D3643: [services] Tunnelbroker - Database item tests initialization.

• karol accepted this revision.Apr 7 2022, 3:45 AM

tomek accepted this revision.Apr 8 2022, 12:30 AM

This revision is now accepted and ready to land.Apr 8 2022, 12:30 AM

Rebase on master.

This revision was landed with ongoing or failed builds.Apr 11 2022, 2:13 AM

Closed by commit rCOMM46e5cc697cdb: [services] Tunnelbroker - Fix `GetN()`->`GetS()`. (authored by • max). · Explain Why

This revision was automatically updated to reflect the committed changes.

• max added a commit: rCOMM46e5cc697cdb: [services] Tunnelbroker - Fix `GetN()`->`GetS()`..

Harbormaster completed remote builds in B8051: Diff 11268.Apr 11 2022, 2:16 AM

Revision Contents
Changeset List

Path

Size

services/

tunnelbroker/

src/

Database/

MessageItem.cpp

2 lines

Diff 11268

View Options

services/tunnelbroker/src/Database/MessageItem.cpp

	#pragma once			#pragma once

	#include "Constants.h"			#include "Constants.h"
	#include "ServerReadReactorBase.h"			#include "ServerReadReactorBase.h"
				#include "ServiceBlobClient.h"
				#include "Tools.h"

	#include "../_generated/backup.grpc.pb.h"			#include "../_generated/backup.grpc.pb.h"
	#include "../_generated/backup.pb.h"			#include "../_generated/backup.pb.h"

	#include <iostream>			#include <iostream>
	#include <memory>			#include <memory>
	#include <string>			#include <string>

	namespace comm {			namespace comm {
	namespace network {			namespace network {
	namespace reactor {			namespace reactor {

	class SendLogReactor : public ServerReadReactorBase<			class SendLogReactor : public ServerReadReactorBase<
	backup::SendLogRequest,			backup::SendLogRequest,
	google::protobuf::Empty> {			google::protobuf::Empty> {
	enum class State {			enum class State {
	USER_ID = 1,			USER_ID = 1,
	LOG_CHUNK = 2,			BACKUP_ID = 2,
				LOG_HASH = 3,
				LOG_CHUNK = 4,
	};			};

	enum class PersistenceMethod {			enum class PersistenceMethod {
	UNKNOWN = 0,			UNKNOWN = 0,
	DB = 1,			DB = 1,
	BLOB = 2,			BLOB = 2,
	};			};

	State state = State::USER_ID;			State state = State::USER_ID;
	PersistenceMethod persistenceMethod = PersistenceMethod::UNKNOWN;			PersistenceMethod persistenceMethod = PersistenceMethod::UNKNOWN;
	std::string userID;			std::string userID;
				std::string backupID;
	void storeInDatabase(const std::string &data) {			std::string hash;
	}			// either the value itself which is a dump of a single operation (if
				// `persistedInBlob` is false) or the holder to blob (if `persistedInBlob` is
	void storeInBlob(const std::string &data) {			// true)
	}			std::string value;
				std::mutex reactorStateMutex;
				std::condition_variable waitingForBlobClientCV;
				std::mutex waitingForBlobClientCVMutex;

				std::shared_ptr<reactor::BlobPutClientReactor> putReactor;
				ServiceBlobClient blobClient;

				void storeInDatabase();
				std::string generateHolder();
				std::string generateLogID();
				void initializePutReactor();

	public:			public:
	using ServerReadReactorBase<backup::SendLogRequest, google::protobuf::Empty>::			using ServerReadReactorBase<backup::SendLogRequest, google::protobuf::Empty>::
	ServerReadReactorBase;			ServerReadReactorBase;

	std::unique_ptr<grpc::Status>			std::unique_ptr<grpc::Status>
	readRequest(backup::SendLogRequest request) override;			readRequest(backup::SendLogRequest request) override;
	void doneCallback() override;			void doneCallback() override;
				void terminateCallback() override;
	};			};

				void SendLogReactor::storeInDatabase() {
				// TODO handle attachment holders
				database::LogItem logItem(
				this->backupID,
				this->generateLogID(),
				(this->persistenceMethod == PersistenceMethod::BLOB),
				this->value,
				{});
				database::DatabaseManager::getInstance().putLogItem(logItem);
				}

				std::string SendLogReactor::generateHolder() {
				// TODO replace mock
				return generateRandomString();
				}

				std::string SendLogReactor::generateLogID() {
				// TODO replace mock
				return generateRandomString();
				}

				void SendLogReactor::initializePutReactor() {
				if (this->value.empty() \|\| this->hash.empty()) {
				throw std::runtime_error("not enough data to initialize put reactor");
				}
				if (this->putReactor == nullptr) {
				this->putReactor = std::make_shared<reactor::BlobPutClientReactor>(
				this->value, this->hash, &this->waitingForBlobClientCV);
				this->blobClient.put(this->putReactor);
				}
				}

	std::unique_ptr<grpc::Status>			std::unique_ptr<grpc::Status>
	SendLogReactor::readRequest(backup::SendLogRequest request) {			SendLogReactor::readRequest(backup::SendLogRequest request) {
				// we make sure that the blob client's state is flushed to the main memory
				// as there may be multiple threads from the pool taking over here
				const std::lock_guard<std::mutex> lock(this->reactorStateMutex);
	switch (this->state) {			switch (this->state) {
	case State::USER_ID: {			case State::USER_ID: {
	if (!request.has_userid()) {			if (!request.has_userid()) {
	throw std::runtime_error("user id expected but not received");			throw std::runtime_error("user id expected but not received");
	}			}
	this->userID = request.userid();			this->userID = request.userid();
				this->state = State::BACKUP_ID;
				return nullptr;
				};
				case State::BACKUP_ID: {
				if (!request.has_backupid()) {
				throw std::runtime_error("backup id expected but not received");
				}
				this->backupID = request.backupid();
				this->state = State::LOG_HASH;
				return nullptr;
				};
				case State::LOG_HASH: {
				if (!request.has_loghash()) {
				throw std::runtime_error("log hash expected but not received");
				}
				this->hash = request.loghash();
	this->state = State::LOG_CHUNK;			this->state = State::LOG_CHUNK;
	return nullptr;			return nullptr;
	};			};
	case State::LOG_CHUNK: {			case State::LOG_CHUNK: {
	if (!request.has_logdata()) {			if (!request.has_logdata()) {
	throw std::runtime_error("log data expected but not received");			throw std::runtime_error("log data expected but not received");
	}			}
	if (this->persistenceMethod == PersistenceMethod::DB) {			std::unique_ptr<std::string> chunk =
	throw std::runtime_error(			std::make_unique<std::string>(std::move(*request.mutable_logdata()));
	"storing multiple chunks in the database is not allowed");			if (chunk->size() == 0) {
				return std::make_unique<grpc::Status>(grpc::Status::OK);
	}			}
	std::string *chunk = request.mutable_logdata();
	// decide if keep in DB or upload to blob			// decide if keep in DB or upload to blob
	if (chunk->size() <= LOG_DATA_SIZE_DATABASE_LIMIT) {			if (chunk->size() <= LOG_DATA_SIZE_DATABASE_LIMIT) {
	if (this->persistenceMethod == PersistenceMethod::UNKNOWN) {			if (this->persistenceMethod == PersistenceMethod::UNKNOWN) {
	this->persistenceMethod = PersistenceMethod::DB;			this->persistenceMethod = PersistenceMethod::DB;
	this->storeInDatabase(*chunk);			this->value = std::move(*chunk);
				this->storeInDatabase();
				return std::make_unique<grpc::Status>(grpc::Status::OK);
	} else if (this->persistenceMethod == PersistenceMethod::BLOB) {			} else if (this->persistenceMethod == PersistenceMethod::BLOB) {
	this->storeInBlob(*chunk);			this->initializePutReactor();
				this->putReactor->scheduleSendingDataChunk(std::move(chunk));
	}			}
	throw std::runtime_error(			throw std::runtime_error(
	"error - invalid persistence state for chunk smaller than database "			"error - invalid persistence state for chunk smaller than database "
	"limit");			"limit");
	} else {			} else {
	if (this->persistenceMethod != PersistenceMethod::UNKNOWN &&			if (this->persistenceMethod != PersistenceMethod::UNKNOWN &&
	this->persistenceMethod != PersistenceMethod::BLOB) {			this->persistenceMethod != PersistenceMethod::BLOB) {
	throw std::runtime_error(			throw std::runtime_error(
	"error - invalid persistence state, uploading to blob should be "			"error - invalid persistence state, uploading to blob should be "
	"continued but it is not");			"continued but it is not");
	}			}
				if (this->persistenceMethod == PersistenceMethod::UNKNOWN) {
	this->persistenceMethod = PersistenceMethod::BLOB;			this->persistenceMethod = PersistenceMethod::BLOB;
	this->storeInBlob(*chunk);
	}			}
	std::cout << "log data received " << chunk->size() << std::endl;			if (this->value.empty()) {
				this->value = this->generateHolder();
				}
				this->initializePutReactor();
				this->putReactor->scheduleSendingDataChunk(std::move(chunk));
				}

	return nullptr;			return nullptr;
	};			};
	}			}
	throw std::runtime_error("send log - invalid state");			throw std::runtime_error("send log - invalid state");
	}			}

				void SendLogReactor::terminateCallback() {
				const std::lock_guard<std::mutex> lock(this->reactorStateMutex);

				if (this->persistenceMethod == PersistenceMethod::DB \|\|
				this->putReactor == nullptr) {
				return;
				}
				this->putReactor->scheduleSendingDataChunk(std::make_unique<std::string>(""));
				std::unique_lock<std::mutex> lock2(this->waitingForBlobClientCVMutex);
				this->waitingForBlobClientCV.wait(lock2);
				// store in db only when we successfully upload chunks
				this->storeInDatabase();
				}

	void SendLogReactor::doneCallback() {			void SendLogReactor::doneCallback() {
				const std::lock_guard<std::mutex> lock(this->reactorStateMutex);
	// TODO implement			// TODO implement
	std::cout << "receive logs done " << this->status.error_code() << "/"			std::cout << "receive logs done " << this->status.error_code() << "/"
	<< this->status.error_message() << std::endl;			<< this->status.error_message() << std::endl;
	}			}

	} // namespace reactor			} // namespace reactor
	} // namespace network			} // namespace network
	} // namespace comm			} // namespace comm

[services] Tunnelbroker - Fix `GetN()`->`GetS()`.ClosedPublicActions