server.cpp.hpp

Go to the documentation of this file.

#include "fl/remote/rpc/server.h"
#include "fl/stl/json.h"
#include "fl/stl/cstddef.h"
#include "fl/stl/move.h"
#include "fl/stl/optional.h"
#include "fl/stl/vector.h"
#include "fl/stl/noexcept.h"
 
namespace fl {
 
Server::Server() FL_NO_EXCEPT
    : mRequestSource([]() { return fl::nullopt; })
    , mResponseSink([](const fl::json&) {})
{}
 
Server::Server(RequestSource source, ResponseSink sink)
    FL_NO_EXCEPT : mRequestSource(fl::move(source))
    , mResponseSink(fl::move(sink))
{}
 
void Server::setRequestHandler(RequestHandler handler) FL_NO_EXCEPT {
    mRequestHandler = fl::move(handler);
}
 
void Server::setRequestSource(RequestSource source) FL_NO_EXCEPT {
    mRequestSource = fl::move(source);
}
 
void Server::setResponseSink(ResponseSink sink) FL_NO_EXCEPT {
    mResponseSink = fl::move(sink);
}
 
void Server::setResponseStreamSink(ResponseStreamSink sink) FL_NO_EXCEPT {
#if FL_PLATFORM_HAS_LARGE_MEMORY
    mResponseStreamSink = fl::move(sink);
#else
    (void)sink;
#endif
}
 
size_t Server::update() FL_NO_EXCEPT {
    size_t processed = pull();
    size_t sent = push();
    return processed + sent;
}
 
size_t Server::pull() FL_NO_EXCEPT {
    if (!mRequestSource || !mRequestHandler) {
        return 0;
    }
 
    size_t processed = 0;
 
    // Pull JSON-RPC requests from source until none available
    while (auto optRequest = mRequestSource()) {
        fl::json request = fl::move(*optRequest);
 
        // Process request through handler
        fl::json response = mRequestHandler(request);
 
        // Queue response (skip scheduled acknowledgments and async skip markers).
        // `noEnqueue` is the new envelope marker (#3228); accept legacy `__skip`
        // for one release of back-compat.
        bool isScheduledAck = response.contains("scheduled") && response["scheduled"].as_bool().value_or(false);
        bool isAsyncSkip = (response.contains("noEnqueue") && response["noEnqueue"].as_bool().value_or(false))
                       || (response.contains("__skip") && response["__skip"].as_bool().value_or(false));
        if (!response.is_null() && !isScheduledAck && !isAsyncSkip) {
            mOutgoingQueue.push_back(fl::move(response));
        }
 
        processed++;
    }
 
    return processed;
}
 
size_t Server::push() FL_NO_EXCEPT {
    if (!mResponseSink) {
        return 0;
    }
 
    size_t sent = 0;
 
    // Push queued responses
    while (!mOutgoingQueue.empty()) {
        mResponseSink(mOutgoingQueue[0]);
        mOutgoingQueue.erase(mOutgoingQueue.begin());
        sent++;
    }
 
    return sent;
}
 
} // namespace fl