diff --git a/services/backup/docker-server/contents/server/CMakeLists.txt b/services/backup/docker-server/contents/server/CMakeLists.txt --- a/services/backup/docker-server/contents/server/CMakeLists.txt +++ b/services/backup/docker-server/contents/server/CMakeLists.txt @@ -50,7 +50,11 @@ include_directories( ./src ./src/DatabaseEntities - ./src/Reactors + ./src/Reactors/server + ./src/Reactors/server/base-reactors + ./src/Reactors/client + ./src/Reactors/client/blob + ./src/Reactors/client/base-reactors ./_generated ${FOLLY_INCLUDES} ./lib/double-conversion diff --git a/services/backup/docker-server/contents/server/src/BackupServiceImpl.cpp b/services/backup/docker-server/contents/server/src/BackupServiceImpl.cpp --- a/services/backup/docker-server/contents/server/src/BackupServiceImpl.cpp +++ b/services/backup/docker-server/contents/server/src/BackupServiceImpl.cpp @@ -1,7 +1,7 @@ #include "BackupServiceImpl.h" -#include "BidiReactorBase.h" -#include "ReadReactorBase.h" +#include "ServerBidiReactorBase.h" +#include "ServerReadReactorBase.h" #include <aws/core/Aws.h> @@ -20,7 +20,7 @@ backup::CreateNewBackupRequest, backup::CreateNewBackupResponse> * BackupServiceImpl::CreateNewBackup(grpc::CallbackServerContext *context) { - class CreateNewBackupReactor : public BidiReactorBase< + class CreateNewBackupReactor : public reactor::ServerBidiReactorBase< backup::CreateNewBackupRequest, backup::CreateNewBackupResponse> { public: @@ -39,12 +39,13 @@ grpc::ServerReadReactor<backup::SendLogRequest> *BackupServiceImpl::SendLog( grpc::CallbackServerContext *context, google::protobuf::Empty *response) { - class SendLogReactor : public ReadReactorBase< + class SendLogReactor : public reactor::ServerReadReactorBase< backup::SendLogRequest, google::protobuf::Empty> { public: - using ReadReactorBase<backup::SendLogRequest, google::protobuf::Empty>:: - ReadReactorBase; + using ServerReadReactorBase< + backup::SendLogRequest, + google::protobuf::Empty>::ServerReadReactorBase; std::unique_ptr<grpc::Status> readRequest(backup::SendLogRequest request) override { // TODO handle request @@ -60,7 +61,7 @@ backup::RecoverBackupKeyRequest, backup::RecoverBackupKeyResponse> * BackupServiceImpl::RecoverBackupKey(grpc::CallbackServerContext *context) { - class RecoverBackupKeyReactor : public BidiReactorBase< + class RecoverBackupKeyReactor : public reactor::ServerBidiReactorBase< backup::RecoverBackupKeyRequest, backup::RecoverBackupKeyResponse> { public: @@ -78,7 +79,7 @@ grpc::ServerBidiReactor<backup::PullBackupRequest, backup::PullBackupResponse> * BackupServiceImpl::PullBackup(grpc::CallbackServerContext *context) { - class PullBackupReactor : public BidiReactorBase< + class PullBackupReactor : public reactor::ServerBidiReactorBase< backup::PullBackupRequest, backup::PullBackupResponse> { public: diff --git a/services/backup/docker-server/contents/server/src/Reactors/BidiReactorBase.h b/services/backup/docker-server/contents/server/src/Reactors/BidiReactorBase.h deleted file mode 100644 --- a/services/backup/docker-server/contents/server/src/Reactors/BidiReactorBase.h +++ /dev/null @@ -1,68 +0,0 @@ -#pragma once - -#include <grpcpp/grpcpp.h> -#include <iostream> -#include <memory> -#include <string> - -namespace comm { -namespace network { - -template <class Request, class Response> -class BidiReactorBase : public grpc::ServerBidiReactor<Request, Response> { - Request request; - Response response; - -public: - BidiReactorBase(); - - void OnDone() override; - void OnReadDone(bool ok) override; - void OnWriteDone(bool ok) override; - - virtual std::unique_ptr<grpc::Status> - handleRequest(Request request, Response *response) = 0; -}; - -template <class Request, class Response> -BidiReactorBase<Request, Response>::BidiReactorBase() { - this->StartRead(&this->request); -} - -template <class Request, class Response> -void BidiReactorBase<Request, Response>::OnDone() { - delete this; -} - -template <class Request, class Response> -void BidiReactorBase<Request, Response>::OnReadDone(bool ok) { - if (!ok) { - this->Finish( - grpc::Status(grpc::StatusCode::INTERNAL, "OnReadDone: reading error")); - return; - } - this->response = Response(); - try { - std::unique_ptr<grpc::Status> status = - this->handleRequest(this->request, &this->response); - if (status != nullptr) { - this->Finish(*status); - return; - } - this->StartWrite(&this->response); - } catch (std::runtime_error &e) { - this->Finish(grpc::Status(grpc::StatusCode::INTERNAL, e.what())); - } -} - -template <class Request, class Response> -void BidiReactorBase<Request, Response>::OnWriteDone(bool ok) { - if (!ok) { - std::cout << "Server write failed" << std::endl; - return; - } - this->StartRead(&this->request); -} - -} // namespace network -} // namespace comm diff --git a/services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerBidiReactorBase.h b/services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerBidiReactorBase.h new file mode 100644 --- /dev/null +++ b/services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerBidiReactorBase.h @@ -0,0 +1,93 @@ +#pragma once + +#include <grpcpp/grpcpp.h> + +#include <iostream> +#include <memory> +#include <string> + +namespace comm { +namespace network { +namespace reactor { + +template <class Request, class Response> +class ServerBidiReactorBase + : public grpc::ServerBidiReactor<Request, Response> { + Request request; + Response response; + +protected: + grpc::Status status; + bool readingAborted = false; + bool sendLastResponse = false; + +public: + ServerBidiReactorBase(); + + void OnDone() override; + void OnReadDone(bool ok) override; + void OnWriteDone(bool ok) override; + + void terminate(grpc::Status status); + + virtual std::unique_ptr<grpc::Status> + handleRequest(Request request, Response *response) = 0; + virtual void initialize(){}; + virtual void doneCallback(){}; +}; + +template <class Request, class Response> +ServerBidiReactorBase<Request, Response>::ServerBidiReactorBase() { + this->initialize(); + this->StartRead(&this->request); +} + +template <class Request, class Response> +void ServerBidiReactorBase<Request, Response>::OnDone() { + this->doneCallback(); + delete this; +} + +template <class Request, class Response> +void ServerBidiReactorBase<Request, Response>::terminate(grpc::Status status) { + this->status = status; + if (this->sendLastResponse) { + this->StartWriteAndFinish(&this->response, grpc::WriteOptions(), status); + } else { + this->Finish(status); + } +} + +template <class Request, class Response> +void ServerBidiReactorBase<Request, Response>::OnReadDone(bool ok) { + if (!ok) { + this->readingAborted = true; + this->terminate(grpc::Status(grpc::StatusCode::ABORTED, "no more reads")); + return; + } + try { + this->response = Response(); + std::unique_ptr<grpc::Status> status = + this->handleRequest(this->request, &this->response); + if (status != nullptr) { + this->terminate(*status); + return; + } + this->StartWrite(&this->response); + } catch (std::runtime_error &e) { + this->terminate(grpc::Status(grpc::StatusCode::INTERNAL, e.what())); + } +} + +template <class Request, class Response> +void ServerBidiReactorBase<Request, Response>::OnWriteDone(bool ok) { + if (!ok) { + std::cout << "Server write failed" << std::endl; + return; + } + this->StartRead(&this->request); +} + +} // namespace reactor +} // namespace network +} // namespace comm diff --git a/services/backup/docker-server/contents/server/src/Reactors/ReadReactorBase.h b/services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerReadReactorBase.h rename from services/backup/docker-server/contents/server/src/Reactors/ReadReactorBase.h rename to services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerReadReactorBase.h --- a/services/backup/docker-server/contents/server/src/Reactors/ReadReactorBase.h +++ b/services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerReadReactorBase.h @@ -1,59 +1,75 @@ #pragma once #include <grpcpp/grpcpp.h> + #include <iostream> #include <memory> #include <string> namespace comm { namespace network { +namespace reactor { template <class Request, class Response> -class ReadReactorBase : public grpc::ServerReadReactor<Request> { +class ServerReadReactorBase : public grpc::ServerReadReactor<Request> { Request request; + void terminate(grpc::Status status); + protected: Response *response; + grpc::Status status; public: - ReadReactorBase(Response *response); + ServerReadReactorBase(Response *response); void OnDone() override; void OnReadDone(bool ok) override; virtual std::unique_ptr<grpc::Status> readRequest(Request request) = 0; + virtual void initialize(){}; + virtual void doneCallback(){}; }; template <class Request, class Response> -ReadReactorBase<Request, Response>::ReadReactorBase(Response *response) +void ServerReadReactorBase<Request, Response>::terminate(grpc::Status status) { + this->status = status; + this->Finish(status); +} + +template <class Request, class Response> +ServerReadReactorBase<Request, Response>::ServerReadReactorBase( + Response *response) : response(response) { + this->initialize(); this->StartRead(&this->request); } template <class Request, class Response> -void ReadReactorBase<Request, Response>::OnDone() { +void ServerReadReactorBase<Request, Response>::OnDone() { + this->doneCallback(); delete this; } template <class Request, class Response> -void ReadReactorBase<Request, Response>::OnReadDone(bool ok) { +void ServerReadReactorBase<Request, Response>::OnReadDone(bool ok) { if (!ok) { - this->Finish( - grpc::Status(grpc::StatusCode::INTERNAL, "OnReadDone: reading error")); + this->terminate(grpc::Status(grpc::StatusCode::INTERNAL, "reading error")); return; } try { std::unique_ptr<grpc::Status> status = this->readRequest(this->request); if (status != nullptr) { - this->Finish(*status); + this->terminate(*status); return; } } catch (std::runtime_error &e) { - this->Finish(grpc::Status(grpc::StatusCode::INTERNAL, e.what())); + this->terminate(grpc::Status(grpc::StatusCode::INTERNAL, e.what())); return; } this->StartRead(&this->request); } +} // namespace reactor } // namespace network } // namespace comm diff --git a/services/backup/docker-server/contents/server/src/Reactors/WriteReactorBase.h b/services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerWriteReactorBase.h rename from services/backup/docker-server/contents/server/src/Reactors/WriteReactorBase.h rename to services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerWriteReactorBase.h --- a/services/backup/docker-server/contents/server/src/Reactors/WriteReactorBase.h +++ b/services/backup/docker-server/contents/server/src/Reactors/server/base-reactors/ServerWriteReactorBase.h @@ -1,33 +1,39 @@ #pragma once #include <grpcpp/grpcpp.h> + #include <iostream> #include <memory> #include <string> namespace comm { namespace network { +namespace reactor { template <class Request, class Response> -class WriteReactorBase : public grpc::ServerWriteReactor<Response> { +class ServerWriteReactorBase : public grpc::ServerWriteReactor<Response> { Response response; + bool initialized = false; protected: // this is a const ref since it's not meant to be modified const Request &request; public: - WriteReactorBase(const Request *request); + ServerWriteReactorBase(const Request *request); virtual void NextWrite(); void OnDone() override; void OnWriteDone(bool ok) override; virtual std::unique_ptr<grpc::Status> writeResponse(Response *response) = 0; + virtual void initialize(){}; + virtual void doneCallback(){}; }; template <class Request, class Response> -WriteReactorBase<Request, Response>::WriteReactorBase(const Request *request) +ServerWriteReactorBase<Request, Response>::ServerWriteReactorBase( + const Request *request) : request(*request) { // we cannot call this->NextWrite() here because it's going to call it on // the base class, not derived leading to the runtime error of calling @@ -37,23 +43,33 @@ } template <class Request, class Response> -void WriteReactorBase<Request, Response>::NextWrite() { - this->response = Response(); - std::unique_ptr<grpc::Status> status = this->writeResponse(&this->response); - if (status != nullptr) { - this->Finish(*status); - return; +void ServerWriteReactorBase<Request, Response>::NextWrite() { + try { + if (!this->initialized) { + this->initialize(); + this->initialized = true; + } + this->response = Response(); + std::unique_ptr<grpc::Status> status = this->writeResponse(&this->response); + if (status != nullptr) { + this->Finish(*status); + return; + } + this->StartWrite(&this->response); + } catch (std::runtime_error &e) { + std::cout << "error: " << e.what() << std::endl; + this->Finish(grpc::Status(grpc::StatusCode::INTERNAL, e.what())); } - this->StartWrite(&this->response); } template <class Request, class Response> -void WriteReactorBase<Request, Response>::OnDone() { +void ServerWriteReactorBase<Request, Response>::OnDone() { + this->doneCallback(); delete this; } template <class Request, class Response> -void WriteReactorBase<Request, Response>::OnWriteDone(bool ok) { +void ServerWriteReactorBase<Request, Response>::OnWriteDone(bool ok) { if (!ok) { this->Finish(grpc::Status(grpc::StatusCode::INTERNAL, "writing error")); return; @@ -65,5 +81,6 @@ } } +} // namespace reactor } // namespace network } // namespace comm