mirror of https://github.com/oxen-io/oxen-mq.git
Provide caller Access in Message
This lets a callback set up something at, say, basic level, but provide different values for an admin auth remote than a basic auth remote.
This commit is contained in:
parent
99bbf8dea9
commit
730633bbae
|
@ -19,16 +19,16 @@ std::ostream& operator<<(std::ostream& os, AuthLevel a);
|
||||||
/// The access level for a command category
|
/// The access level for a command category
|
||||||
struct Access {
|
struct Access {
|
||||||
/// Minimum access level required
|
/// Minimum access level required
|
||||||
AuthLevel auth = AuthLevel::none;
|
AuthLevel auth;
|
||||||
/// If true only remote SNs may call the category commands
|
/// If true only remote SNs may call the category commands
|
||||||
bool remote_sn = false;
|
bool remote_sn;
|
||||||
/// If true the category requires that the local node is a SN
|
/// If true the category requires that the local node is a SN
|
||||||
bool local_sn = false;
|
bool local_sn;
|
||||||
|
|
||||||
/// Constructor. Intentionally allows implicit conversion from an AuthLevel so that an
|
/// Constructor. Intentionally allows implicit conversion from an AuthLevel so that an
|
||||||
/// AuthLevel can be passed anywhere an Access is required (the resulting Access will have both
|
/// AuthLevel can be passed anywhere an Access is required (the resulting Access will have both
|
||||||
/// remote and local sn set to false).
|
/// remote and local sn set to false).
|
||||||
Access(AuthLevel auth, bool remote_sn = false, bool local_sn = false)
|
Access(AuthLevel auth = AuthLevel::none, bool remote_sn = false, bool local_sn = false)
|
||||||
: auth{auth}, remote_sn{remote_sn}, local_sn{local_sn} {}
|
: auth{auth}, remote_sn{remote_sn}, local_sn{local_sn} {}
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|
|
@ -341,20 +341,21 @@ void LokiMQ::set_general_threads(int threads) {
|
||||||
general_workers = threads;
|
general_workers = threads;
|
||||||
}
|
}
|
||||||
|
|
||||||
LokiMQ::run_info& LokiMQ::run_info::load(category* cat_, std::string command_, ConnectionID conn_,
|
LokiMQ::run_info& LokiMQ::run_info::load(category* cat_, std::string command_, ConnectionID conn_, Access access_,
|
||||||
std::vector<zmq::message_t> data_parts_, const std::pair<CommandCallback, bool>* callback_) {
|
std::vector<zmq::message_t> data_parts_, const std::pair<CommandCallback, bool>* callback_) {
|
||||||
is_batch_job = false;
|
is_batch_job = false;
|
||||||
is_reply_job = false;
|
is_reply_job = false;
|
||||||
cat = cat_;
|
cat = cat_;
|
||||||
command = std::move(command_);
|
command = std::move(command_);
|
||||||
conn = std::move(conn_);
|
conn = std::move(conn_);
|
||||||
|
access = std::move(access_);
|
||||||
data_parts = std::move(data_parts_);
|
data_parts = std::move(data_parts_);
|
||||||
callback = callback_;
|
callback = callback_;
|
||||||
return *this;
|
return *this;
|
||||||
}
|
}
|
||||||
|
|
||||||
LokiMQ::run_info& LokiMQ::run_info::load(pending_command&& pending) {
|
LokiMQ::run_info& LokiMQ::run_info::load(pending_command&& pending) {
|
||||||
return load(&pending.cat, std::move(pending.command), std::move(pending.conn),
|
return load(&pending.cat, std::move(pending.command), std::move(pending.conn), std::move(pending.access),
|
||||||
std::move(pending.data_parts), pending.callback);
|
std::move(pending.data_parts), pending.callback);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -586,11 +586,12 @@ private:
|
||||||
std::vector<zmq::message_t> data_parts;
|
std::vector<zmq::message_t> data_parts;
|
||||||
const std::pair<CommandCallback, bool>* callback;
|
const std::pair<CommandCallback, bool>* callback;
|
||||||
ConnectionID conn;
|
ConnectionID conn;
|
||||||
|
Access access;
|
||||||
|
|
||||||
pending_command(category& cat, std::string command, std::vector<zmq::message_t> data_parts,
|
pending_command(category& cat, std::string command, std::vector<zmq::message_t> data_parts,
|
||||||
const std::pair<CommandCallback, bool>* callback, ConnectionID conn)
|
const std::pair<CommandCallback, bool>* callback, ConnectionID conn, Access access)
|
||||||
: cat{cat}, command{std::move(command)}, data_parts{std::move(data_parts)},
|
: cat{cat}, command{std::move(command)}, data_parts{std::move(data_parts)},
|
||||||
callback{callback}, conn{std::move(conn)} {}
|
callback{callback}, conn{std::move(conn)}, access{std::move(access)} {}
|
||||||
};
|
};
|
||||||
std::list<pending_command> pending_commands;
|
std::list<pending_command> pending_commands;
|
||||||
|
|
||||||
|
@ -610,6 +611,7 @@ private:
|
||||||
category *cat;
|
category *cat;
|
||||||
std::string command;
|
std::string command;
|
||||||
ConnectionID conn; // The connection (or SN pubkey) to reply on/to.
|
ConnectionID conn; // The connection (or SN pubkey) to reply on/to.
|
||||||
|
Access access; // The access level of the invoker (actual level, can be higher than the command's requirement)
|
||||||
std::string conn_route; // if non-empty this is the reply routing prefix (for incoming connections)
|
std::string conn_route; // if non-empty this is the reply routing prefix (for incoming connections)
|
||||||
std::vector<zmq::message_t> data_parts;
|
std::vector<zmq::message_t> data_parts;
|
||||||
|
|
||||||
|
@ -627,7 +629,7 @@ private:
|
||||||
std::string worker_routing_id; // "w123" where 123 == worker_id
|
std::string worker_routing_id; // "w123" where 123 == worker_id
|
||||||
|
|
||||||
/// Loads the run info with an incoming command
|
/// Loads the run info with an incoming command
|
||||||
run_info& load(category* cat, std::string command, ConnectionID conn,
|
run_info& load(category* cat, std::string command, ConnectionID conn, Access access,
|
||||||
std::vector<zmq::message_t> data_parts, const std::pair<CommandCallback, bool>* callback);
|
std::vector<zmq::message_t> data_parts, const std::pair<CommandCallback, bool>* callback);
|
||||||
|
|
||||||
/// Loads the run info with a stored pending command
|
/// Loads the run info with a stored pending command
|
||||||
|
|
|
@ -15,9 +15,10 @@ public:
|
||||||
std::vector<string_view> data; ///< The provided command data parts, if any.
|
std::vector<string_view> data; ///< The provided command data parts, if any.
|
||||||
ConnectionID conn; ///< The connection info for routing a reply; also contains the pubkey/sn status.
|
ConnectionID conn; ///< The connection info for routing a reply; also contains the pubkey/sn status.
|
||||||
std::string reply_tag; ///< If the invoked command is a request command this is the required reply tag that will be prepended by `send_reply()`.
|
std::string reply_tag; ///< If the invoked command is a request command this is the required reply tag that will be prepended by `send_reply()`.
|
||||||
|
Access access; ///< The access level of the invoker. This can be higher than the access level of the command, for example for an admin invoking a basic command.
|
||||||
|
|
||||||
/// Constructor
|
/// Constructor
|
||||||
Message(LokiMQ& lmq, ConnectionID cid) : lokimq{lmq}, conn{std::move(cid)} {}
|
Message(LokiMQ& lmq, ConnectionID cid, Access access) : lokimq{lmq}, conn{std::move(cid)}, access{std::move(access)} {}
|
||||||
|
|
||||||
// Non-copyable
|
// Non-copyable
|
||||||
Message(const Message&) = delete;
|
Message(const Message&) = delete;
|
||||||
|
|
|
@ -21,7 +21,7 @@ void LokiMQ::worker_thread(unsigned int index) {
|
||||||
LMQ_LOG(debug, "New worker thread ", worker_id, " started");
|
LMQ_LOG(debug, "New worker thread ", worker_id, " started");
|
||||||
sock.connect(SN_ADDR_WORKERS);
|
sock.connect(SN_ADDR_WORKERS);
|
||||||
|
|
||||||
Message message{*this, 0};
|
Message message{*this, 0, AuthLevel::none};
|
||||||
std::vector<zmq::message_t> parts;
|
std::vector<zmq::message_t> parts;
|
||||||
run_info& run = workers[index]; // This contains our first job, and will be updated later with subsequent jobs
|
run_info& run = workers[index]; // This contains our first job, and will be updated later with subsequent jobs
|
||||||
|
|
||||||
|
@ -37,6 +37,7 @@ void LokiMQ::worker_thread(unsigned int index) {
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
message.conn = run.conn;
|
message.conn = run.conn;
|
||||||
|
message.access = run.access;
|
||||||
message.data.clear();
|
message.data.clear();
|
||||||
|
|
||||||
LMQ_TRACE("Got incoming command from ", message.conn, message.conn.route.empty() ? "(outgoing)" : "(incoming)");
|
LMQ_TRACE("Got incoming command from ", message.conn, message.conn.route.empty() ? "(outgoing)" : "(incoming)");
|
||||||
|
@ -254,6 +255,7 @@ void LokiMQ::proxy_to_worker(size_t conn_index, std::vector<zmq::message_t>& par
|
||||||
return;
|
return;
|
||||||
|
|
||||||
auto& category = *cat_call.first;
|
auto& category = *cat_call.first;
|
||||||
|
Access access{peer->auth_level, peer->service_node, local_service_node};
|
||||||
|
|
||||||
if (category.active_threads >= category.reserved_threads && active_workers() >= general_workers) {
|
if (category.active_threads >= category.reserved_threads && active_workers() >= general_workers) {
|
||||||
// No free reserved or general spots, try to queue it for later
|
// No free reserved or general spots, try to queue it for later
|
||||||
|
@ -265,7 +267,7 @@ void LokiMQ::proxy_to_worker(size_t conn_index, std::vector<zmq::message_t>& par
|
||||||
|
|
||||||
LMQ_LOG(debug, "No available free workers, queuing ", command, " for later");
|
LMQ_LOG(debug, "No available free workers, queuing ", command, " for later");
|
||||||
ConnectionID conn{peer->service_node ? ConnectionID::SN_ID : conn_index_to_id[conn_index].id, peer->pubkey, std::move(tmp_peer.route)};
|
ConnectionID conn{peer->service_node ? ConnectionID::SN_ID : conn_index_to_id[conn_index].id, peer->pubkey, std::move(tmp_peer.route)};
|
||||||
pending_commands.emplace_back(category, std::move(command), std::move(data_parts), cat_call.second, std::move(conn));
|
pending_commands.emplace_back(category, std::move(command), std::move(data_parts), cat_call.second, std::move(conn), std::move(access));
|
||||||
category.queued++;
|
category.queued++;
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
@ -281,7 +283,7 @@ void LokiMQ::proxy_to_worker(size_t conn_index, std::vector<zmq::message_t>& par
|
||||||
c.route = std::move(tmp_peer.route);
|
c.route = std::move(tmp_peer.route);
|
||||||
if (outgoing || peer->service_node)
|
if (outgoing || peer->service_node)
|
||||||
tmp_peer.route.clear();
|
tmp_peer.route.clear();
|
||||||
run.load(&category, std::move(command), std::move(c), std::move(data_parts), cat_call.second);
|
run.load(&category, std::move(command), std::move(c), std::move(access), std::move(data_parts), cat_call.second);
|
||||||
}
|
}
|
||||||
|
|
||||||
if (outgoing)
|
if (outgoing)
|
||||||
|
|
Loading…
Reference in New Issue