Mirror/zeek
1
0
Fork 0
mirror of https://github.com/zeek/zeek.git synced 2025-10-15 21:18:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

broker: Make MakeHub() filter_type to empty, adapt WebSocketShim

Browse source 
... also removes handling for topics that are never subscribed to anymore.
This commit is contained in:
Arne Welzel 2025-10-15 09:40:35 +02:00
parent 97d30cfec1
commit 41fca08e66
4 changed files with 5 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/broker/Manager.cc
View file

@ -2132,9 +2132,9 @@ void Manager::PrepareForwarding(const std::string& name) {
DBG_LOG(DBG_BROKER, "Resolved table forward for data store %s", name.c_str());
}
broker::hub Manager::MakeHub(broker::filter_type ft) {
broker::hub Manager::MakeHub() {
++hub_count;
return bstate->endpoint.make_hub(std::move(ft));
return bstate->endpoint.make_hub({});
}
void Manager::DestroyHub(broker::hub&& hub) { --hub_count; }

2
src/broker/Manager.h
View file

@ -455,7 +455,7 @@ private:
friend class WebSocketState;
// Create a hub for WebSocket clients.
broker::hub MakeHub(broker::filter_type ft);
broker::hub MakeHub();
// This hub is to be destroyed.
void DestroyHub(broker::hub&& hub);

11
src/broker/WebSocketShim.cc
View file

@ -22,7 +22,7 @@ namespace zeek::Broker {
class WebSocketState {
public:
WebSocketState() : hub(broker_mgr->MakeHub({broker::topic::errors()})) {}
WebSocketState() : hub(broker_mgr->MakeHub()) {}
~WebSocketState() {
// Let the manager know we're done with this hook.
broker_mgr->DestroyHub(std::move(hub));
@ -154,17 +154,8 @@ void WebSocketShim::Process() {
BROKER_WS_DEBUG("Shim: Process() got %zu messages (%s)", messages.size(), NodeId().c_str());
for ( auto& message : messages ) {
auto&& topic = broker::get_topic(message);
if ( broker::is_prefix(topic, broker::topic::errors_str) ) {
std::string err_msg = broker::to_string(message);
ProcessError("broker_error", broker::to_string(err_msg));
}
else if ( broker::is_prefix(topic, broker::topic::statuses_str) ) {
// Ignore status messages for WebSocket clients
}
else {
broker::zeek::visit_as_message([this, topic](auto& msg) { ProcessMessage(topic, msg); }, message);
}
}
}
void WebSocketShim::ProcessMessage(std::string_view topic, broker::zeek::Batch& batch) {

2
testing/btest/Baseline/cluster.websocket.broker.spurious-peer-error/client.out
View file

@ -1,7 +1,5 @@
### BTest baseline data generated by btest-diff. Do not edit. Use "btest -U/-u" to update. Requires BTest >= 0.63.
Connected!
{"client": "ws1", "error": "KeyError('data')", "data": {"type": "error", "code": "broker_error", "context": "data(<$>/local/data/errors, (error, peer_unavailable, ((nil, 127.0.0.1, 21/tcp, 1), unable to connect to remote peer)))"}}
{"client": "ws1", "error": "KeyError('data')", "data": {"type": "error", "code": "broker_error", "context": "data(<$>/local/data/errors, (error, peer_unavailable, ((nil, 127.0.0.1, 21/tcp, 1), unable to connect to remote peer)))"}}
{"client": "ws1", "topic": "/test/pings/reply", "event_name": "ping", "event_args": [{"@data-type": "string", "data": "ws1-1"}, {"@data-type": "count", "data": 101}]}
{"client": "ws1", "topic": "/test/pings/reply", "event_name": "ping", "event_args": [{"@data-type": "string", "data": "ws1-2"}, {"@data-type": "count", "data": 102}]}
{"client": "ws1", "topic": "/test/pings/reply", "event_name": "ping", "event_args": [{"@data-type": "string", "data": "ws1-3"}, {"@data-type": "count", "data": 103}]}