#pragma once #include "wfmo.h" #include "stat.h" #include #include #include #include #include namespace NNeh { struct TMessage { TMessage() = default; inline TMessage(TString addr, TString data) : Addr(std::move(addr)) , Data(std::move(data)) { } static TMessage FromString(TStringBuf request); TString Addr; TString Data; }; using TMessageRef = TAutoPtr; struct TError { public: enum TType { UnknownType, Cancelled, ProtocolSpecific }; TError(TString text, TType type = UnknownType, i32 code = 0, i32 systemCode = 0) : Type(std::move(type)) , Code(code) , Text(text) , SystemCode(systemCode) { } TType Type = UnknownType; i32 Code = 0; // protocol specific code (example(http): 404) TString Text; i32 SystemCode = 0; // system error code }; using TErrorRef = TAutoPtr; struct TResponse; using TResponseRef = TAutoPtr; struct TResponse { inline TResponse(TMessage req, TString data, const TDuration duration) : TResponse(std::move(req), std::move(data), duration, {} /* firstLine */, {} /* headers */, {} /* error */) { } inline TResponse(TMessage req, TString data, const TDuration duration, TString firstLine, THttpHeaders headers) : TResponse(std::move(req), std::move(data), duration, std::move(firstLine), std::move(headers), {} /* error */) { } inline TResponse(TMessage req, TString data, const TDuration duration, TString firstLine, THttpHeaders headers, TErrorRef error) : Request(std::move(req)) , Data(std::move(data)) , Duration(duration) , FirstLine(std::move(firstLine)) , Headers(std::move(headers)) , Error_(std::move(error)) { } inline static TResponseRef FromErrorText(TMessage msg, TString error, const TDuration duration) { return new TResponse(std::move(msg), {} /* data */, duration, {} /* firstLine */, {} /* headers */, new TError(std::move(error))); } inline static TResponseRef FromError(TMessage msg, TErrorRef error, const TDuration duration) { return new TResponse(std::move(msg), {} /* data */, duration, {} /* firstLine */, {} /* headers */, error); } inline static TResponseRef FromError(TMessage msg, TErrorRef error, const TDuration duration, TString data, TString firstLine, THttpHeaders headers) { return new TResponse(std::move(msg), std::move(data), duration, std::move(firstLine), std::move(headers), error); } inline static TResponseRef FromError( TMessage msg, TErrorRef error, TString data, const TDuration duration, TString firstLine, THttpHeaders headers) { return new TResponse(std::move(msg), std::move(data), duration, std::move(firstLine), std::move(headers), error); } inline bool IsError() const { return Error_.Get(); } inline TError::TType GetErrorType() const { return Error_.Get() ? Error_->Type : TError::UnknownType; } inline i32 GetErrorCode() const { return Error_.Get() ? Error_->Code : 0; } inline i32 GetSystemErrorCode() const { return Error_.Get() ? Error_->SystemCode : 0; } inline TString GetErrorText() const { return Error_.Get() ? Error_->Text : TString(); } const TMessage Request; const TString Data; const TDuration Duration; const TString FirstLine; THttpHeaders Headers; private: THolder Error_; }; class THandle; class IOnRecv { public: virtual ~IOnRecv() = default; virtual void OnNotify(THandle&) { } //callback on receive response virtual void OnEnd() { } //response was extracted by Wait() method, - OnRecv() will not be called virtual void OnRecv(THandle& resp) = 0; //callback on destroy handler }; class THandle: public TThrRefBase, public TWaitHandle { public: inline THandle(IOnRecv* f, TStatCollector* s = nullptr) noexcept : F_(f) , Stat_(s) { } ~THandle() override { if (F_) { try { F_->OnRecv(*this); } catch (...) { } } } virtual bool MessageSendedCompletely() const noexcept { //TODO return true; } virtual void Cancel() noexcept { //TODO if (!!Stat_) Stat_->OnCancel(); } inline const TResponse* Response() const noexcept { return R_.Get(); } //method MUST be called only after success Wait() for this handle or from callback IOnRecv::OnRecv() //else exist chance for memory leak (race between Get()/Notify()) inline TResponseRef Get() noexcept { return R_; } inline bool Wait(TResponseRef& msg, const TInstant deadLine) { if (WaitForOne(*this, deadLine)) { if (F_) { F_->OnEnd(); F_ = nullptr; } msg = Get(); return true; } return false; } inline bool Wait(TResponseRef& msg, const TDuration timeOut) { return Wait(msg, timeOut.ToDeadLine()); } inline bool Wait(TResponseRef& msg) { return Wait(msg, TInstant::Max()); } inline TResponseRef Wait(const TInstant deadLine) { TResponseRef ret; Wait(ret, deadLine); return ret; } inline TResponseRef Wait(const TDuration timeOut) { return Wait(timeOut.ToDeadLine()); } inline TResponseRef Wait() { return Wait(TInstant::Max()); } protected: inline void Notify(TResponseRef resp) { if (!!Stat_) { if (!resp || resp->IsError()) { Stat_->OnFail(); } else { Stat_->OnSuccess(); } } R_.Swap(resp); if (F_) { try { F_->OnNotify(*this); } catch (...) { } } Signal(); } IOnRecv* F_; private: TResponseRef R_; THolder Stat_; }; using THandleRef = TIntrusivePtr; THandleRef Request(const TMessage& msg, IOnRecv* fallback); inline THandleRef Request(const TMessage& msg) { return Request(msg, nullptr); } THandleRef Request(const TString& req, IOnRecv* fallback); inline THandleRef Request(const TString& req) { return Request(req, nullptr); } class IMultiRequester { public: virtual ~IMultiRequester() = default; virtual void Add(const THandleRef& req) = 0; virtual void Del(const THandleRef& req) = 0; virtual bool Wait(THandleRef& req, TInstant deadLine) = 0; virtual bool IsEmpty() const = 0; inline void Schedule(const TString& req) { Add(Request(req)); } inline bool Wait(THandleRef& req, TDuration timeOut) { return Wait(req, timeOut.ToDeadLine()); } inline bool Wait(THandleRef& req) { return Wait(req, TInstant::Max()); } inline bool Wait(TResponseRef& resp, TInstant deadLine) { THandleRef req; while (Wait(req, deadLine)) { resp = req->Get(); if (!!resp) { return true; } } return false; } inline bool Wait(TResponseRef& resp) { return Wait(resp, TInstant::Max()); } }; using IMultiRequesterRef = TAutoPtr; IMultiRequesterRef CreateRequester(); bool SetProtocolOption(TStringBuf protoOption, TStringBuf value); }