Skip to content

Wire broker readiness notifications#992

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Weidong Cui (wdcui) merged 15 commits into from
Jul 2, 2026
Merged

Wire broker readiness notifications #992

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
b011ebf
Wire broker readiness notifications
wdcui Jul 1, 2026
1f2007e
Clarify broker host channel generic names
wdcui Jul 1, 2026
f4c14a2
Require broker notification channel
wdcui Jul 1, 2026
3b6cd95
Rename paired broker serving API
wdcui Jul 1, 2026
a6ae87d
Align broker host channel error bounds
wdcui Jul 1, 2026
f43f2f6
Attach notifier to broker host connection
wdcui Jul 1, 2026
8825534
Emit readiness notifications from event handling
wdcui Jul 1, 2026
f85dc25
Avoid duplicate readiness notifications
wdcui Jul 1, 2026
c37ef5a
Exercise readiness responses through serve_connection
wdcui Jul 1, 2026
8c1e075
Inline broker host request loop
wdcui Jul 1, 2026
3eb6ec5
Rename broker control socket argument
wdcui Jul 1, 2026
7f9b074
Clarify notification receiver thread ownership
wdcui Jul 1, 2026
9a185f3
Simplify broker socket option handling
wdcui Jul 1, 2026
5597b37
Require broker socket paths in connect
wdcui Jul 1, 2026
73f09bc
Merge broker socket retry helpers
wdcui Jul 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
152 changes: 121 additions & 31 deletions litebox_broker_host/src/lib.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -14,7 +14,9 @@ extern crate std;

use litebox_broker_core::{BrokerCore, BrokerSession, CallerCredential};
use litebox_broker_protocol::BROKER_PROTOCOL_VERSION;
use litebox_broker_protocol::channel::{HostControlChannel, HostReceive, PeerCredential};
use litebox_broker_protocol::channel::{
HostControlChannel, HostNotificationChannel, HostReceive, PeerCredential,
};
use litebox_broker_protocol::error::ErrorCode;
use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse, WaitEventResponse};
use litebox_broker_protocol::message::{
Expand All@@ -25,15 +27,24 @@ mod error;

pub use error::{BrokerHostError, Result};

/// Authenticates, negotiates, and serves one broker connection over the control channel.
pub fn serve_connection<Channel>(
/// Authenticates, negotiates, and serves one broker association over paired
/// control and notification channels.
///
/// The deployment must bind both channels to the same authenticated peer
/// association. Active requests and responses remain on the control channel;
/// broker-initiated readiness wakeups are sent on the notification channel.
/// Event mutations caused by control requests return readiness in their control
/// response and do not also emit a duplicate notification.
pub fn serve_connection<ControlChannel, NotificationChannel, ChannelError>(
core: &BrokerCore,
channel: &mut Channel,
) -> Result<ConnectionTermination, Channel::Error>
control_channel: &mut ControlChannel,
_notification_channel: &mut NotificationChannel,
) -> Result<ConnectionTermination, ChannelError>
where
Channel: HostControlChannel,
ControlChannel: HostControlChannel<Error = ChannelError>,
NotificationChannel: HostNotificationChannel<Error = ChannelError>,
{
let peer_credential = channel
let peer_credential = control_channel
.peer_credential()
.map_err(BrokerHostError::Channel)?;
let caller_credential = match peer_credential {
Expand All@@ -43,13 +54,13 @@ where
let session = core.create_session(caller_credential)?;

loop {
let request = match channel
let request = match control_channel
.recv_handshake_request()
.map_err(BrokerHostError::Channel)?
{
HostReceive::Message(request) => request,
HostReceive::ProtocolViolation => {
channel
control_channel
.send_handshake_response(&BrokerHandshakeResponse::Error(
ErrorCode::ProtocolState,
))
Expand All@@ -69,38 +80,31 @@ where
broker_protocol_version: BROKER_PROTOCOL_VERSION,
}
};
channel
control_channel
.send_handshake_response(&response)
.map_err(BrokerHostError::Channel)?;
if negotiated {
break;
}
}

serve_request_loop(channel, &session)
}

fn serve_request_loop<Channel>(
channel: &mut Channel,
session: &BrokerSession,
) -> Result<ConnectionTermination, Channel::Error>
where
Channel: HostControlChannel,
{
loop {
let request = match channel.recv_request().map_err(BrokerHostError::Channel)? {
let request = match control_channel
.recv_request()
.map_err(BrokerHostError::Channel)?
{
HostReceive::Message(request) => request,
HostReceive::ProtocolViolation => {
channel
control_channel
.send_response(&BrokerResponse::Error(ErrorCode::ProtocolState))
.map_err(BrokerHostError::Channel)?;
return Ok(ConnectionTermination::ProtocolViolation);
}
HostReceive::PeerClosed => break,
};

let response = handle_request(session, request);
channel
let response = handle_request(&session, request);
control_channel
.send_response(&response)
.map_err(BrokerHostError::Channel)?;
}
Expand DownExpand Up@@ -167,8 +171,11 @@ pub enum ConnectionTermination {
mod tests {
use super::*;
use litebox_broker_core::{PolicyEngine, PrincipalRights};
use litebox_broker_protocol::event::{CreateEventRequest, WaitEventRequest};
use litebox_broker_protocol::message::BrokerHandshakeRequest;
use litebox_broker_protocol::event::{
AddEventRequest, ConsumeEventRequest, CreateEventRequest, EventConsumeMode,
WaitEventRequest,
};
use litebox_broker_protocol::message::{BrokerHandshakeRequest, BrokerNotification};
use litebox_broker_protocol::{ObjectHandle, ProtocolVersion};

#[test]
Expand All@@ -183,6 +190,7 @@ mod tests {
serve_connection_rejects_active_request_before_negotiation(&broker);
serve_connection_rejects_handshake_request_after_negotiation(&broker);
serve_connection_returns_channel_error_when_response_send_fails(&broker);
serve_connection_returns_event_readiness_in_control_responses(&broker);
active_request_closes_object_reference(&broker);
}

Expand All@@ -198,9 +206,10 @@ mod tests {
Ok(HostReceive::PeerClosed),
]),
);
let mut notifications = FakeHostNotificationChannel::default();

assert_eq!(
serve_connection(broker, &mut channel).unwrap(),
serve_connection(broker, &mut channel, &mut notifications).unwrap(),
ConnectionTermination::PeerClosed
);
assert_eq!(
Expand DownExpand Up@@ -228,9 +237,10 @@ mod tests {
]),
std::vec::Vec::from([Ok(HostReceive::PeerClosed)]),
);
let mut notifications = FakeHostNotificationChannel::default();

assert_eq!(
serve_connection(broker, &mut channel).unwrap(),
serve_connection(broker, &mut channel, &mut notifications).unwrap(),
ConnectionTermination::PeerClosed
);
assert_eq!(
Expand All@@ -251,9 +261,10 @@ mod tests {
std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
std::vec::Vec::new(),
);
let mut notifications = FakeHostNotificationChannel::default();

assert_eq!(
serve_connection(broker, &mut channel).unwrap(),
serve_connection(broker, &mut channel, &mut notifications).unwrap(),
ConnectionTermination::ProtocolViolation
);
assert_eq!(
Expand All@@ -270,9 +281,10 @@ mod tests {
}))]),
std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
);
let mut notifications = FakeHostNotificationChannel::default();

assert_eq!(
serve_connection(broker, &mut channel).unwrap(),
serve_connection(broker, &mut channel, &mut notifications).unwrap(),
ConnectionTermination::ProtocolViolation
);
assert_eq!(
Expand All@@ -295,14 +307,54 @@ mod tests {
std::vec::Vec::new(),
);
channel.send_error = true;
let mut notifications = FakeHostNotificationChannel::default();

match serve_connection(broker, &mut channel) {
match serve_connection(broker, &mut channel, &mut notifications) {
Err(BrokerHostError::Channel(())) => {}
result => panic!("unexpected serve result: {result:?}"),
}
assert!(channel.handshake_responses.is_empty());
}

fn serve_connection_returns_event_readiness_in_control_responses(broker: &BrokerCore) {
let mut channel = FakeHostControlChannel::new(
std::vec::Vec::from([Ok(HostReceive::Message(BrokerHandshakeRequest {
protocol_version: BROKER_PROTOCOL_VERSION,
}))]),
std::vec::Vec::from([Ok(HostReceive::Message(BrokerRequest::Event(
EventRequest::Create(CreateEventRequest { initial_count: 0 }),
)))]),
);
channel.enqueue_readiness_requests_after_create = true;
let mut notifications = FakeHostNotificationChannel::default();

assert_eq!(
serve_connection(broker, &mut channel, &mut notifications).unwrap(),
ConnectionTermination::PeerClosed
);
assert!(notifications.notifications.is_empty());
assert_eq!(
&channel.responses[1..],
[
BrokerResponse::Event(EventResponse::Add(AddEventResponse {
readiness: litebox_broker_protocol::event::ReadinessState {
read_ready: true,
write_ready: true,
},
})),
BrokerResponse::Event(EventResponse::Consume(
litebox_broker_protocol::event::ConsumeEventResponse {
value: 1,
readiness: litebox_broker_protocol::event::ReadinessState {
read_ready: false,
write_ready: true,
},
}
)),
]
);
}

fn active_request_closes_object_reference(broker: &BrokerCore) {
let session = broker
.create_session(CallerCredential::Unauthenticated)
Expand DownExpand Up@@ -344,6 +396,7 @@ mod tests {
requests: std::vec::Vec<core::result::Result<HostReceive<BrokerRequest>, ()>>,
handshake_responses: std::vec::Vec<BrokerHandshakeResponse>,
responses: std::vec::Vec<BrokerResponse>,
enqueue_readiness_requests_after_create: bool,
send_error: bool,
}

Expand All@@ -359,6 +412,7 @@ mod tests {
requests,
handshake_responses: std::vec::Vec::new(),
responses: std::vec::Vec::new(),
enqueue_readiness_requests_after_create: false,
send_error: false,
}
}
Expand DownExpand Up@@ -409,8 +463,44 @@ mod tests {
if self.send_error {
return Err(());
}
if self.enqueue_readiness_requests_after_create
&& let BrokerResponse::Event(EventResponse::Create(response)) = response
{
self.requests
.push(Ok(HostReceive::Message(BrokerRequest::Event(
EventRequest::Add(AddEventRequest {
handle: response.handle,
value: 1,
}),
))));
self.requests
.push(Ok(HostReceive::Message(BrokerRequest::Event(
EventRequest::Consume(ConsumeEventRequest {
handle: response.handle,
mode: EventConsumeMode::One,
}),
))));
self.requests.push(Ok(HostReceive::PeerClosed));
}
self.responses.push(response.clone());
Ok(())
}
}

#[derive(Default)]
struct FakeHostNotificationChannel {
notifications: std::vec::Vec<BrokerNotification>,
}

impl HostNotificationChannel for FakeHostNotificationChannel {
type Error = ();

fn send_notification(
&mut self,
notification: &BrokerNotification,
) -> core::result::Result<(), Self::Error> {
self.notifications.push(notification.clone());
Ok(())
}
}
}
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
Wire broker readiness notifications by wdcui · Pull Request #992 · microsoft/litebox · GitHub
Skip to content
Wire broker readiness notifications#992

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Weidong Cui (wdcui)
merged 15 commits into

from

Jul 2, 2026
 Merged


Wire broker readiness notifications
#992
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


15 commits


Select commit
Hold shift + click to select a range
b011ebf

Wire broker readiness notifications


wdcui Jul 1, 2026
1f2007e

Clarify broker host channel generic names


wdcui Jul 1, 2026
f4c14a2

Require broker notification channel


wdcui Jul 1, 2026
3b6cd95

Rename paired broker serving API


wdcui Jul 1, 2026
a6ae87d

Align broker host channel error bounds


wdcui Jul 1, 2026
f43f2f6

Attach notifier to broker host connection


wdcui Jul 1, 2026
8825534

Emit readiness notifications from event handling


wdcui Jul 1, 2026
f85dc25

Avoid duplicate readiness notifications


wdcui Jul 1, 2026
c37ef5a

Exercise readiness responses through serve_connection


wdcui Jul 1, 2026
8c1e075

Inline broker host request loop


wdcui Jul 1, 2026
3eb6ec5

Rename broker control socket argument


wdcui Jul 1, 2026
7f9b074

Clarify notification receiver thread ownership


wdcui Jul 1, 2026
9a185f3

Simplify broker socket option handling


wdcui Jul 1, 2026
5597b37

Require broker socket paths in connect


wdcui Jul 1, 2026
73f09bc

Merge broker socket retry helpers


wdcui Jul 1, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


152 changes: 121 additions & 31 deletions
litebox_broker_host/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -14,7 +14,9 @@ extern crate std;

use litebox_broker_core::{BrokerCore, BrokerSession, CallerCredential};
use litebox_broker_protocol::BROKER_PROTOCOL_VERSION;
use litebox_broker_protocol::channel::{HostControlChannel, HostReceive, PeerCredential};
use litebox_broker_protocol::channel::{
 HostControlChannel, HostNotificationChannel, HostReceive, PeerCredential,
};
use litebox_broker_protocol::error::ErrorCode;
use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse, WaitEventResponse};
use litebox_broker_protocol::message::{
Expand All@@ -25,15 +27,24 @@ mod error;

pub use error::{BrokerHostError, Result};

/// Authenticates, negotiates, and serves one broker connection over the control channel.
pub fn serve_connection<Channel>(
/// Authenticates, negotiates, and serves one broker association over paired
/// control and notification channels.
///
/// The deployment must bind both channels to the same authenticated peer
/// association. Active requests and responses remain on the control channel;
/// broker-initiated readiness wakeups are sent on the notification channel.
/// Event mutations caused by control requests return readiness in their control
/// response and do not also emit a duplicate notification.
pub fn serve_connection<ControlChannel, NotificationChannel, ChannelError>(
 core: &BrokerCore,
 channel: &mut Channel,
) -> Result<ConnectionTermination, Channel::Error>
 control_channel: &mut ControlChannel,
 _notification_channel: &mut NotificationChannel,
) -> Result<ConnectionTermination, ChannelError>
where
 Channel: HostControlChannel,
 ControlChannel: HostControlChannel<Error = ChannelError>,
 NotificationChannel: HostNotificationChannel<Error = ChannelError>,
{
 let peer_credential = channel
 let peer_credential = control_channel
 .peer_credential()
 .map_err(BrokerHostError::Channel)?;
 let caller_credential = match peer_credential {
Expand All@@ -43,13 +54,13 @@ where
 let session = core.create_session(caller_credential)?;

 loop {
 let request = match channel
 let request = match control_channel
 .recv_handshake_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_handshake_response(&BrokerHandshakeResponse::Error(
 ErrorCode::ProtocolState,
 ))
Expand All@@ -69,38 +80,31 @@ where
 broker_protocol_version: BROKER_PROTOCOL_VERSION,
 }
 };
channel
control_channel
 .send_handshake_response(&response)
 .map_err(BrokerHostError::Channel)?;
 if negotiated {
 break;
 }
 }

 serve_request_loop(channel, &session)
}

fn serve_request_loop<Channel>(
 channel: &mut Channel,
 session: &BrokerSession,
) -> Result<ConnectionTermination, Channel::Error>
where
 Channel: HostControlChannel,
{
 loop {
 let request = match channel.recv_request().map_err(BrokerHostError::Channel)? {
 let request = match control_channel
 .recv_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_response(&BrokerResponse::Error(ErrorCode::ProtocolState))
 .map_err(BrokerHostError::Channel)?;
 return Ok(ConnectionTermination::ProtocolViolation);
 }
 HostReceive::PeerClosed => break,
 };

 let response = handle_request(session, request);
channel
 let response = handle_request(&session, request);
control_channel
 .send_response(&response)
 .map_err(BrokerHostError::Channel)?;
 }
Expand DownExpand Up@@ -167,8 +171,11 @@ pub enum ConnectionTermination {
mod tests {
 use super::*;
 use litebox_broker_core::{PolicyEngine, PrincipalRights};
 use litebox_broker_protocol::event::{CreateEventRequest, WaitEventRequest};
 use litebox_broker_protocol::message::BrokerHandshakeRequest;
 use litebox_broker_protocol::event::{
 AddEventRequest, ConsumeEventRequest, CreateEventRequest, EventConsumeMode,
 WaitEventRequest,
 };
 use litebox_broker_protocol::message::{BrokerHandshakeRequest, BrokerNotification};
 use litebox_broker_protocol::{ObjectHandle, ProtocolVersion};

 #[test]
Expand All@@ -183,6 +190,7 @@ mod tests {
 serve_connection_rejects_active_request_before_negotiation(&broker);
 serve_connection_rejects_handshake_request_after_negotiation(&broker);
 serve_connection_returns_channel_error_when_response_send_fails(&broker);
 serve_connection_returns_event_readiness_in_control_responses(&broker);
 active_request_closes_object_reference(&broker);
 }

Expand All@@ -198,9 +206,10 @@ mod tests {
 Ok(HostReceive::PeerClosed),
 ]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand DownExpand Up@@ -228,9 +237,10 @@ mod tests {
 ]),
 std::vec::Vec::from([Ok(HostReceive::PeerClosed)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand All@@ -251,9 +261,10 @@ mod tests {
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 std::vec::Vec::new(),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -270,9 +281,10 @@ mod tests {
 }))]),
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -295,14 +307,54 @@ mod tests {
 std::vec::Vec::new(),
 );
 channel.send_error = true;
 let mut notifications = FakeHostNotificationChannel::default();

 match serve_connection(broker, &mut channel) {
 match serve_connection(broker, &mut channel, &mut notifications) {
 Err(BrokerHostError::Channel(())) => {}
 result => panic!("unexpected serve result: {result:?}"),
 }
 assert!(channel.handshake_responses.is_empty());
 }

 fn serve_connection_returns_event_readiness_in_control_responses(broker: &BrokerCore) {
 let mut channel = FakeHostControlChannel::new(
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerHandshakeRequest {
 protocol_version: BROKER_PROTOCOL_VERSION,
 }))]),
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Create(CreateEventRequest { initial_count: 0 }),
 )))]),
 );
 channel.enqueue_readiness_requests_after_create = true;
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert!(notifications.notifications.is_empty());
 assert_eq!(
 &channel.responses[1..],
 [
 BrokerResponse::Event(EventResponse::Add(AddEventResponse {
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: true,
 write_ready: true,
 },
 })),
 BrokerResponse::Event(EventResponse::Consume(
 litebox_broker_protocol::event::ConsumeEventResponse {
 value: 1,
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: false,
 write_ready: true,
 },
 }
 )),
 ]
 );
 }

 fn active_request_closes_object_reference(broker: &BrokerCore) {
 let session = broker
 .create_session(CallerCredential::Unauthenticated)
Expand DownExpand Up@@ -344,6 +396,7 @@ mod tests {
 requests: std::vec::Vec<core::result::Result<HostReceive<BrokerRequest>, ()>>,
 handshake_responses: std::vec::Vec<BrokerHandshakeResponse>,
 responses: std::vec::Vec<BrokerResponse>,
 enqueue_readiness_requests_after_create: bool,
 send_error: bool,
 }

Expand All@@ -359,6 +412,7 @@ mod tests {
 requests,
 handshake_responses: std::vec::Vec::new(),
 responses: std::vec::Vec::new(),
 enqueue_readiness_requests_after_create: false,
 send_error: false,
 }
 }
Expand DownExpand Up@@ -409,8 +463,44 @@ mod tests {
 if self.send_error {
 return Err(());
 }
 if self.enqueue_readiness_requests_after_create
 && let BrokerResponse::Event(EventResponse::Create(response)) = response
 {
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Add(AddEventRequest {
 handle: response.handle,
 value: 1,
 }),
 ))));
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Consume(ConsumeEventRequest {
 handle: response.handle,
 mode: EventConsumeMode::One,
 }),
 ))));
 self.requests.push(Ok(HostReceive::PeerClosed));
 }
 self.responses.push(response.clone());
 Ok(())
 }
 }

 #[derive(Default)]
 struct FakeHostNotificationChannel {
 notifications: std::vec::Vec<BrokerNotification>,
 }

 impl HostNotificationChannel for FakeHostNotificationChannel {
 type Error = ();

 fn send_notification(
 &mut self,
 notification: &BrokerNotification,
 ) -> core::result::Result<(), Self::Error> {
 self.notifications.push(notification.clone());
 Ok(())
 }
 }
}
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Wire broker readiness notifications by wdcui · Pull Request #992 · microsoft/litebox · GitHub
Skip to content
Wire broker readiness notifications#992

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Weidong Cui (wdcui)
merged 15 commits into

from

Jul 2, 2026
 Merged


Wire broker readiness notifications
#992
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


15 commits


Select commit
Hold shift + click to select a range
b011ebf

Wire broker readiness notifications


wdcui Jul 1, 2026
1f2007e

Clarify broker host channel generic names


wdcui Jul 1, 2026
f4c14a2

Require broker notification channel


wdcui Jul 1, 2026
3b6cd95

Rename paired broker serving API


wdcui Jul 1, 2026
a6ae87d

Align broker host channel error bounds


wdcui Jul 1, 2026
f43f2f6

Attach notifier to broker host connection


wdcui Jul 1, 2026
8825534

Emit readiness notifications from event handling


wdcui Jul 1, 2026
f85dc25

Avoid duplicate readiness notifications


wdcui Jul 1, 2026
c37ef5a

Exercise readiness responses through serve_connection


wdcui Jul 1, 2026
8c1e075

Inline broker host request loop


wdcui Jul 1, 2026
3eb6ec5

Rename broker control socket argument


wdcui Jul 1, 2026
7f9b074

Clarify notification receiver thread ownership


wdcui Jul 1, 2026
9a185f3

Simplify broker socket option handling


wdcui Jul 1, 2026
5597b37

Require broker socket paths in connect


wdcui Jul 1, 2026
73f09bc

Merge broker socket retry helpers


wdcui Jul 1, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


152 changes: 121 additions & 31 deletions
litebox_broker_host/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -14,7 +14,9 @@ extern crate std;

use litebox_broker_core::{BrokerCore, BrokerSession, CallerCredential};
use litebox_broker_protocol::BROKER_PROTOCOL_VERSION;
use litebox_broker_protocol::channel::{HostControlChannel, HostReceive, PeerCredential};
use litebox_broker_protocol::channel::{
 HostControlChannel, HostNotificationChannel, HostReceive, PeerCredential,
};
use litebox_broker_protocol::error::ErrorCode;
use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse, WaitEventResponse};
use litebox_broker_protocol::message::{
Expand All@@ -25,15 +27,24 @@ mod error;

pub use error::{BrokerHostError, Result};

/// Authenticates, negotiates, and serves one broker connection over the control channel.
pub fn serve_connection<Channel>(
/// Authenticates, negotiates, and serves one broker association over paired
/// control and notification channels.
///
/// The deployment must bind both channels to the same authenticated peer
/// association. Active requests and responses remain on the control channel;
/// broker-initiated readiness wakeups are sent on the notification channel.
/// Event mutations caused by control requests return readiness in their control
/// response and do not also emit a duplicate notification.
pub fn serve_connection<ControlChannel, NotificationChannel, ChannelError>(
 core: &BrokerCore,
 channel: &mut Channel,
) -> Result<ConnectionTermination, Channel::Error>
 control_channel: &mut ControlChannel,
 _notification_channel: &mut NotificationChannel,
) -> Result<ConnectionTermination, ChannelError>
where
 Channel: HostControlChannel,
 ControlChannel: HostControlChannel<Error = ChannelError>,
 NotificationChannel: HostNotificationChannel<Error = ChannelError>,
{
 let peer_credential = channel
 let peer_credential = control_channel
 .peer_credential()
 .map_err(BrokerHostError::Channel)?;
 let caller_credential = match peer_credential {
Expand All@@ -43,13 +54,13 @@ where
 let session = core.create_session(caller_credential)?;

 loop {
 let request = match channel
 let request = match control_channel
 .recv_handshake_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_handshake_response(&BrokerHandshakeResponse::Error(
 ErrorCode::ProtocolState,
 ))
Expand All@@ -69,38 +80,31 @@ where
 broker_protocol_version: BROKER_PROTOCOL_VERSION,
 }
 };
channel
control_channel
 .send_handshake_response(&response)
 .map_err(BrokerHostError::Channel)?;
 if negotiated {
 break;
 }
 }

 serve_request_loop(channel, &session)
}

fn serve_request_loop<Channel>(
 channel: &mut Channel,
 session: &BrokerSession,
) -> Result<ConnectionTermination, Channel::Error>
where
 Channel: HostControlChannel,
{
 loop {
 let request = match channel.recv_request().map_err(BrokerHostError::Channel)? {
 let request = match control_channel
 .recv_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_response(&BrokerResponse::Error(ErrorCode::ProtocolState))
 .map_err(BrokerHostError::Channel)?;
 return Ok(ConnectionTermination::ProtocolViolation);
 }
 HostReceive::PeerClosed => break,
 };

 let response = handle_request(session, request);
channel
 let response = handle_request(&session, request);
control_channel
 .send_response(&response)
 .map_err(BrokerHostError::Channel)?;
 }
Expand DownExpand Up@@ -167,8 +171,11 @@ pub enum ConnectionTermination {
mod tests {
 use super::*;
 use litebox_broker_core::{PolicyEngine, PrincipalRights};
 use litebox_broker_protocol::event::{CreateEventRequest, WaitEventRequest};
 use litebox_broker_protocol::message::BrokerHandshakeRequest;
 use litebox_broker_protocol::event::{
 AddEventRequest, ConsumeEventRequest, CreateEventRequest, EventConsumeMode,
 WaitEventRequest,
 };
 use litebox_broker_protocol::message::{BrokerHandshakeRequest, BrokerNotification};
 use litebox_broker_protocol::{ObjectHandle, ProtocolVersion};

 #[test]
Expand All@@ -183,6 +190,7 @@ mod tests {
 serve_connection_rejects_active_request_before_negotiation(&broker);
 serve_connection_rejects_handshake_request_after_negotiation(&broker);
 serve_connection_returns_channel_error_when_response_send_fails(&broker);
 serve_connection_returns_event_readiness_in_control_responses(&broker);
 active_request_closes_object_reference(&broker);
 }

Expand All@@ -198,9 +206,10 @@ mod tests {
 Ok(HostReceive::PeerClosed),
 ]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand DownExpand Up@@ -228,9 +237,10 @@ mod tests {
 ]),
 std::vec::Vec::from([Ok(HostReceive::PeerClosed)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand All@@ -251,9 +261,10 @@ mod tests {
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 std::vec::Vec::new(),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -270,9 +281,10 @@ mod tests {
 }))]),
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -295,14 +307,54 @@ mod tests {
 std::vec::Vec::new(),
 );
 channel.send_error = true;
 let mut notifications = FakeHostNotificationChannel::default();

 match serve_connection(broker, &mut channel) {
 match serve_connection(broker, &mut channel, &mut notifications) {
 Err(BrokerHostError::Channel(())) => {}
 result => panic!("unexpected serve result: {result:?}"),
 }
 assert!(channel.handshake_responses.is_empty());
 }

 fn serve_connection_returns_event_readiness_in_control_responses(broker: &BrokerCore) {
 let mut channel = FakeHostControlChannel::new(
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerHandshakeRequest {
 protocol_version: BROKER_PROTOCOL_VERSION,
 }))]),
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Create(CreateEventRequest { initial_count: 0 }),
 )))]),
 );
 channel.enqueue_readiness_requests_after_create = true;
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert!(notifications.notifications.is_empty());
 assert_eq!(
 &channel.responses[1..],
 [
 BrokerResponse::Event(EventResponse::Add(AddEventResponse {
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: true,
 write_ready: true,
 },
 })),
 BrokerResponse::Event(EventResponse::Consume(
 litebox_broker_protocol::event::ConsumeEventResponse {
 value: 1,
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: false,
 write_ready: true,
 },
 }
 )),
 ]
 );
 }

 fn active_request_closes_object_reference(broker: &BrokerCore) {
 let session = broker
 .create_session(CallerCredential::Unauthenticated)
Expand DownExpand Up@@ -344,6 +396,7 @@ mod tests {
 requests: std::vec::Vec<core::result::Result<HostReceive<BrokerRequest>, ()>>,
 handshake_responses: std::vec::Vec<BrokerHandshakeResponse>,
 responses: std::vec::Vec<BrokerResponse>,
 enqueue_readiness_requests_after_create: bool,
 send_error: bool,
 }

Expand All@@ -359,6 +412,7 @@ mod tests {
 requests,
 handshake_responses: std::vec::Vec::new(),
 responses: std::vec::Vec::new(),
 enqueue_readiness_requests_after_create: false,
 send_error: false,
 }
 }
Expand DownExpand Up@@ -409,8 +463,44 @@ mod tests {
 if self.send_error {
 return Err(());
 }
 if self.enqueue_readiness_requests_after_create
 && let BrokerResponse::Event(EventResponse::Create(response)) = response
 {
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Add(AddEventRequest {
 handle: response.handle,
 value: 1,
 }),
 ))));
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Consume(ConsumeEventRequest {
 handle: response.handle,
 mode: EventConsumeMode::One,
 }),
 ))));
 self.requests.push(Ok(HostReceive::PeerClosed));
 }
 self.responses.push(response.clone());
 Ok(())
 }
 }

 #[derive(Default)]
 struct FakeHostNotificationChannel {
 notifications: std::vec::Vec<BrokerNotification>,
 }

 impl HostNotificationChannel for FakeHostNotificationChannel {
 type Error = ();

 fn send_notification(
 &mut self,
 notification: &BrokerNotification,
 ) -> core::result::Result<(), Self::Error> {
 self.notifications.push(notification.clone());
 Ok(())
 }
 }
}
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Wire broker readiness notifications by wdcui · Pull Request #992 · microsoft/litebox · GitHub
Skip to content
Wire broker readiness notifications#992

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Weidong Cui (wdcui)
merged 15 commits into

from

Jul 2, 2026
 Merged


Wire broker readiness notifications
#992
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


15 commits


Select commit
Hold shift + click to select a range
b011ebf

Wire broker readiness notifications


wdcui Jul 1, 2026
1f2007e

Clarify broker host channel generic names


wdcui Jul 1, 2026
f4c14a2

Require broker notification channel


wdcui Jul 1, 2026
3b6cd95

Rename paired broker serving API


wdcui Jul 1, 2026
a6ae87d

Align broker host channel error bounds


wdcui Jul 1, 2026
f43f2f6

Attach notifier to broker host connection


wdcui Jul 1, 2026
8825534

Emit readiness notifications from event handling


wdcui Jul 1, 2026
f85dc25

Avoid duplicate readiness notifications


wdcui Jul 1, 2026
c37ef5a

Exercise readiness responses through serve_connection


wdcui Jul 1, 2026
8c1e075

Inline broker host request loop


wdcui Jul 1, 2026
3eb6ec5

Rename broker control socket argument


wdcui Jul 1, 2026
7f9b074

Clarify notification receiver thread ownership


wdcui Jul 1, 2026
9a185f3

Simplify broker socket option handling


wdcui Jul 1, 2026
5597b37

Require broker socket paths in connect


wdcui Jul 1, 2026
73f09bc

Merge broker socket retry helpers


wdcui Jul 1, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


152 changes: 121 additions & 31 deletions
litebox_broker_host/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -14,7 +14,9 @@ extern crate std;

use litebox_broker_core::{BrokerCore, BrokerSession, CallerCredential};
use litebox_broker_protocol::BROKER_PROTOCOL_VERSION;
use litebox_broker_protocol::channel::{HostControlChannel, HostReceive, PeerCredential};
use litebox_broker_protocol::channel::{
 HostControlChannel, HostNotificationChannel, HostReceive, PeerCredential,
};
use litebox_broker_protocol::error::ErrorCode;
use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse, WaitEventResponse};
use litebox_broker_protocol::message::{
Expand All@@ -25,15 +27,24 @@ mod error;

pub use error::{BrokerHostError, Result};

/// Authenticates, negotiates, and serves one broker connection over the control channel.
pub fn serve_connection<Channel>(
/// Authenticates, negotiates, and serves one broker association over paired
/// control and notification channels.
///
/// The deployment must bind both channels to the same authenticated peer
/// association. Active requests and responses remain on the control channel;
/// broker-initiated readiness wakeups are sent on the notification channel.
/// Event mutations caused by control requests return readiness in their control
/// response and do not also emit a duplicate notification.
pub fn serve_connection<ControlChannel, NotificationChannel, ChannelError>(
 core: &BrokerCore,
 channel: &mut Channel,
) -> Result<ConnectionTermination, Channel::Error>
 control_channel: &mut ControlChannel,
 _notification_channel: &mut NotificationChannel,
) -> Result<ConnectionTermination, ChannelError>
where
 Channel: HostControlChannel,
 ControlChannel: HostControlChannel<Error = ChannelError>,
 NotificationChannel: HostNotificationChannel<Error = ChannelError>,
{
 let peer_credential = channel
 let peer_credential = control_channel
 .peer_credential()
 .map_err(BrokerHostError::Channel)?;
 let caller_credential = match peer_credential {
Expand All@@ -43,13 +54,13 @@ where
 let session = core.create_session(caller_credential)?;

 loop {
 let request = match channel
 let request = match control_channel
 .recv_handshake_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_handshake_response(&BrokerHandshakeResponse::Error(
 ErrorCode::ProtocolState,
 ))
Expand All@@ -69,38 +80,31 @@ where
 broker_protocol_version: BROKER_PROTOCOL_VERSION,
 }
 };
channel
control_channel
 .send_handshake_response(&response)
 .map_err(BrokerHostError::Channel)?;
 if negotiated {
 break;
 }
 }

 serve_request_loop(channel, &session)
}

fn serve_request_loop<Channel>(
 channel: &mut Channel,
 session: &BrokerSession,
) -> Result<ConnectionTermination, Channel::Error>
where
 Channel: HostControlChannel,
{
 loop {
 let request = match channel.recv_request().map_err(BrokerHostError::Channel)? {
 let request = match control_channel
 .recv_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_response(&BrokerResponse::Error(ErrorCode::ProtocolState))
 .map_err(BrokerHostError::Channel)?;
 return Ok(ConnectionTermination::ProtocolViolation);
 }
 HostReceive::PeerClosed => break,
 };

 let response = handle_request(session, request);
channel
 let response = handle_request(&session, request);
control_channel
 .send_response(&response)
 .map_err(BrokerHostError::Channel)?;
 }
Expand DownExpand Up@@ -167,8 +171,11 @@ pub enum ConnectionTermination {
mod tests {
 use super::*;
 use litebox_broker_core::{PolicyEngine, PrincipalRights};
 use litebox_broker_protocol::event::{CreateEventRequest, WaitEventRequest};
 use litebox_broker_protocol::message::BrokerHandshakeRequest;
 use litebox_broker_protocol::event::{
 AddEventRequest, ConsumeEventRequest, CreateEventRequest, EventConsumeMode,
 WaitEventRequest,
 };
 use litebox_broker_protocol::message::{BrokerHandshakeRequest, BrokerNotification};
 use litebox_broker_protocol::{ObjectHandle, ProtocolVersion};

 #[test]
Expand All@@ -183,6 +190,7 @@ mod tests {
 serve_connection_rejects_active_request_before_negotiation(&broker);
 serve_connection_rejects_handshake_request_after_negotiation(&broker);
 serve_connection_returns_channel_error_when_response_send_fails(&broker);
 serve_connection_returns_event_readiness_in_control_responses(&broker);
 active_request_closes_object_reference(&broker);
 }

Expand All@@ -198,9 +206,10 @@ mod tests {
 Ok(HostReceive::PeerClosed),
 ]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand DownExpand Up@@ -228,9 +237,10 @@ mod tests {
 ]),
 std::vec::Vec::from([Ok(HostReceive::PeerClosed)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand All@@ -251,9 +261,10 @@ mod tests {
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 std::vec::Vec::new(),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -270,9 +281,10 @@ mod tests {
 }))]),
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -295,14 +307,54 @@ mod tests {
 std::vec::Vec::new(),
 );
 channel.send_error = true;
 let mut notifications = FakeHostNotificationChannel::default();

 match serve_connection(broker, &mut channel) {
 match serve_connection(broker, &mut channel, &mut notifications) {
 Err(BrokerHostError::Channel(())) => {}
 result => panic!("unexpected serve result: {result:?}"),
 }
 assert!(channel.handshake_responses.is_empty());
 }

 fn serve_connection_returns_event_readiness_in_control_responses(broker: &BrokerCore) {
 let mut channel = FakeHostControlChannel::new(
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerHandshakeRequest {
 protocol_version: BROKER_PROTOCOL_VERSION,
 }))]),
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Create(CreateEventRequest { initial_count: 0 }),
 )))]),
 );
 channel.enqueue_readiness_requests_after_create = true;
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert!(notifications.notifications.is_empty());
 assert_eq!(
 &channel.responses[1..],
 [
 BrokerResponse::Event(EventResponse::Add(AddEventResponse {
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: true,
 write_ready: true,
 },
 })),
 BrokerResponse::Event(EventResponse::Consume(
 litebox_broker_protocol::event::ConsumeEventResponse {
 value: 1,
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: false,
 write_ready: true,
 },
 }
 )),
 ]
 );
 }

 fn active_request_closes_object_reference(broker: &BrokerCore) {
 let session = broker
 .create_session(CallerCredential::Unauthenticated)
Expand DownExpand Up@@ -344,6 +396,7 @@ mod tests {
 requests: std::vec::Vec<core::result::Result<HostReceive<BrokerRequest>, ()>>,
 handshake_responses: std::vec::Vec<BrokerHandshakeResponse>,
 responses: std::vec::Vec<BrokerResponse>,
 enqueue_readiness_requests_after_create: bool,
 send_error: bool,
 }

Expand All@@ -359,6 +412,7 @@ mod tests {
 requests,
 handshake_responses: std::vec::Vec::new(),
 responses: std::vec::Vec::new(),
 enqueue_readiness_requests_after_create: false,
 send_error: false,
 }
 }
Expand DownExpand Up@@ -409,8 +463,44 @@ mod tests {
 if self.send_error {
 return Err(());
 }
 if self.enqueue_readiness_requests_after_create
 && let BrokerResponse::Event(EventResponse::Create(response)) = response
 {
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Add(AddEventRequest {
 handle: response.handle,
 value: 1,
 }),
 ))));
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Consume(ConsumeEventRequest {
 handle: response.handle,
 mode: EventConsumeMode::One,
 }),
 ))));
 self.requests.push(Ok(HostReceive::PeerClosed));
 }
 self.responses.push(response.clone());
 Ok(())
 }
 }

 #[derive(Default)]
 struct FakeHostNotificationChannel {
 notifications: std::vec::Vec<BrokerNotification>,
 }

 impl HostNotificationChannel for FakeHostNotificationChannel {
 type Error = ();

 fn send_notification(
 &mut self,
 notification: &BrokerNotification,
 ) -> core::result::Result<(), Self::Error> {
 self.notifications.push(notification.clone());
 Ok(())
 }
 }
}
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '
Wire broker readiness notifications by wdcui · Pull Request #992 · microsoft/litebox · GitHub
Skip to content
Wire broker readiness notifications#992

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Weidong Cui (wdcui)
merged 15 commits into

from

Jul 2, 2026
 Merged


Wire broker readiness notifications
#992
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


15 commits


Select commit
Hold shift + click to select a range
b011ebf

Wire broker readiness notifications


wdcui Jul 1, 2026
1f2007e

Clarify broker host channel generic names


wdcui Jul 1, 2026
f4c14a2

Require broker notification channel


wdcui Jul 1, 2026
3b6cd95

Rename paired broker serving API


wdcui Jul 1, 2026
a6ae87d

Align broker host channel error bounds


wdcui Jul 1, 2026
f43f2f6

Attach notifier to broker host connection


wdcui Jul 1, 2026
8825534

Emit readiness notifications from event handling


wdcui Jul 1, 2026
f85dc25

Avoid duplicate readiness notifications


wdcui Jul 1, 2026
c37ef5a

Exercise readiness responses through serve_connection


wdcui Jul 1, 2026
8c1e075

Inline broker host request loop


wdcui Jul 1, 2026
3eb6ec5

Rename broker control socket argument


wdcui Jul 1, 2026
7f9b074

Clarify notification receiver thread ownership


wdcui Jul 1, 2026
9a185f3

Simplify broker socket option handling


wdcui Jul 1, 2026
5597b37

Require broker socket paths in connect


wdcui Jul 1, 2026
73f09bc

Merge broker socket retry helpers


wdcui Jul 1, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


152 changes: 121 additions & 31 deletions
litebox_broker_host/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -14,7 +14,9 @@ extern crate std;

use litebox_broker_core::{BrokerCore, BrokerSession, CallerCredential};
use litebox_broker_protocol::BROKER_PROTOCOL_VERSION;
use litebox_broker_protocol::channel::{HostControlChannel, HostReceive, PeerCredential};
use litebox_broker_protocol::channel::{
 HostControlChannel, HostNotificationChannel, HostReceive, PeerCredential,
};
use litebox_broker_protocol::error::ErrorCode;
use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse, WaitEventResponse};
use litebox_broker_protocol::message::{
Expand All@@ -25,15 +27,24 @@ mod error;

pub use error::{BrokerHostError, Result};

/// Authenticates, negotiates, and serves one broker connection over the control channel.
pub fn serve_connection<Channel>(
/// Authenticates, negotiates, and serves one broker association over paired
/// control and notification channels.
///
/// The deployment must bind both channels to the same authenticated peer
/// association. Active requests and responses remain on the control channel;
/// broker-initiated readiness wakeups are sent on the notification channel.
/// Event mutations caused by control requests return readiness in their control
/// response and do not also emit a duplicate notification.
pub fn serve_connection<ControlChannel, NotificationChannel, ChannelError>(
 core: &BrokerCore,
 channel: &mut Channel,
) -> Result<ConnectionTermination, Channel::Error>
 control_channel: &mut ControlChannel,
 _notification_channel: &mut NotificationChannel,
) -> Result<ConnectionTermination, ChannelError>
where
 Channel: HostControlChannel,
 ControlChannel: HostControlChannel<Error = ChannelError>,
 NotificationChannel: HostNotificationChannel<Error = ChannelError>,
{
 let peer_credential = channel
 let peer_credential = control_channel
 .peer_credential()
 .map_err(BrokerHostError::Channel)?;
 let caller_credential = match peer_credential {
Expand All@@ -43,13 +54,13 @@ where
 let session = core.create_session(caller_credential)?;

 loop {
 let request = match channel
 let request = match control_channel
 .recv_handshake_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_handshake_response(&BrokerHandshakeResponse::Error(
 ErrorCode::ProtocolState,
 ))
Expand All@@ -69,38 +80,31 @@ where
 broker_protocol_version: BROKER_PROTOCOL_VERSION,
 }
 };
channel
control_channel
 .send_handshake_response(&response)
 .map_err(BrokerHostError::Channel)?;
 if negotiated {
 break;
 }
 }

 serve_request_loop(channel, &session)
}

fn serve_request_loop<Channel>(
 channel: &mut Channel,
 session: &BrokerSession,
) -> Result<ConnectionTermination, Channel::Error>
where
 Channel: HostControlChannel,
{
 loop {
 let request = match channel.recv_request().map_err(BrokerHostError::Channel)? {
 let request = match control_channel
 .recv_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_response(&BrokerResponse::Error(ErrorCode::ProtocolState))
 .map_err(BrokerHostError::Channel)?;
 return Ok(ConnectionTermination::ProtocolViolation);
 }
 HostReceive::PeerClosed => break,
 };

 let response = handle_request(session, request);
channel
 let response = handle_request(&session, request);
control_channel
 .send_response(&response)
 .map_err(BrokerHostError::Channel)?;
 }
Expand DownExpand Up@@ -167,8 +171,11 @@ pub enum ConnectionTermination {
mod tests {
 use super::*;
 use litebox_broker_core::{PolicyEngine, PrincipalRights};
 use litebox_broker_protocol::event::{CreateEventRequest, WaitEventRequest};
 use litebox_broker_protocol::message::BrokerHandshakeRequest;
 use litebox_broker_protocol::event::{
 AddEventRequest, ConsumeEventRequest, CreateEventRequest, EventConsumeMode,
 WaitEventRequest,
 };
 use litebox_broker_protocol::message::{BrokerHandshakeRequest, BrokerNotification};
 use litebox_broker_protocol::{ObjectHandle, ProtocolVersion};

 #[test]
Expand All@@ -183,6 +190,7 @@ mod tests {
 serve_connection_rejects_active_request_before_negotiation(&broker);
 serve_connection_rejects_handshake_request_after_negotiation(&broker);
 serve_connection_returns_channel_error_when_response_send_fails(&broker);
 serve_connection_returns_event_readiness_in_control_responses(&broker);
 active_request_closes_object_reference(&broker);
 }

Expand All@@ -198,9 +206,10 @@ mod tests {
 Ok(HostReceive::PeerClosed),
 ]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand DownExpand Up@@ -228,9 +237,10 @@ mod tests {
 ]),
 std::vec::Vec::from([Ok(HostReceive::PeerClosed)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand All@@ -251,9 +261,10 @@ mod tests {
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 std::vec::Vec::new(),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -270,9 +281,10 @@ mod tests {
 }))]),
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -295,14 +307,54 @@ mod tests {
 std::vec::Vec::new(),
 );
 channel.send_error = true;
 let mut notifications = FakeHostNotificationChannel::default();

 match serve_connection(broker, &mut channel) {
 match serve_connection(broker, &mut channel, &mut notifications) {
 Err(BrokerHostError::Channel(())) => {}
 result => panic!("unexpected serve result: {result:?}"),
 }
 assert!(channel.handshake_responses.is_empty());
 }

 fn serve_connection_returns_event_readiness_in_control_responses(broker: &BrokerCore) {
 let mut channel = FakeHostControlChannel::new(
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerHandshakeRequest {
 protocol_version: BROKER_PROTOCOL_VERSION,
 }))]),
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Create(CreateEventRequest { initial_count: 0 }),
 )))]),
 );
 channel.enqueue_readiness_requests_after_create = true;
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert!(notifications.notifications.is_empty());
 assert_eq!(
 &channel.responses[1..],
 [
 BrokerResponse::Event(EventResponse::Add(AddEventResponse {
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: true,
 write_ready: true,
 },
 })),
 BrokerResponse::Event(EventResponse::Consume(
 litebox_broker_protocol::event::ConsumeEventResponse {
 value: 1,
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: false,
 write_ready: true,
 },
 }
 )),
 ]
 );
 }

 fn active_request_closes_object_reference(broker: &BrokerCore) {
 let session = broker
 .create_session(CallerCredential::Unauthenticated)
Expand DownExpand Up@@ -344,6 +396,7 @@ mod tests {
 requests: std::vec::Vec<core::result::Result<HostReceive<BrokerRequest>, ()>>,
 handshake_responses: std::vec::Vec<BrokerHandshakeResponse>,
 responses: std::vec::Vec<BrokerResponse>,
 enqueue_readiness_requests_after_create: bool,
 send_error: bool,
 }

Expand All@@ -359,6 +412,7 @@ mod tests {
 requests,
 handshake_responses: std::vec::Vec::new(),
 responses: std::vec::Vec::new(),
 enqueue_readiness_requests_after_create: false,
 send_error: false,
 }
 }
Expand DownExpand Up@@ -409,8 +463,44 @@ mod tests {
 if self.send_error {
 return Err(());
 }
 if self.enqueue_readiness_requests_after_create
 && let BrokerResponse::Event(EventResponse::Create(response)) = response
 {
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Add(AddEventRequest {
 handle: response.handle,
 value: 1,
 }),
 ))));
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Consume(ConsumeEventRequest {
 handle: response.handle,
 mode: EventConsumeMode::One,
 }),
 ))));
 self.requests.push(Ok(HostReceive::PeerClosed));
 }
 self.responses.push(response.clone());
 Ok(())
 }
 }

 #[derive(Default)]
 struct FakeHostNotificationChannel {
 notifications: std::vec::Vec<BrokerNotification>,
 }

 impl HostNotificationChannel for FakeHostNotificationChannel {
 type Error = ();

 fn send_notification(
 &mut self,
 notification: &BrokerNotification,
 ) -> core::result::Result<(), Self::Error> {
 self.notifications.push(notification.clone());
 Ok(())
 }
 }
}
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Wire broker readiness notifications by wdcui · Pull Request #992 · microsoft/litebox · GitHub
Skip to content
Wire broker readiness notifications#992

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Weidong Cui (wdcui)
merged 15 commits into

from

Jul 2, 2026
 Merged


Wire broker readiness notifications
#992
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


15 commits


Select commit
Hold shift + click to select a range
b011ebf

Wire broker readiness notifications


wdcui Jul 1, 2026
1f2007e

Clarify broker host channel generic names


wdcui Jul 1, 2026
f4c14a2

Require broker notification channel


wdcui Jul 1, 2026
3b6cd95

Rename paired broker serving API


wdcui Jul 1, 2026
a6ae87d

Align broker host channel error bounds


wdcui Jul 1, 2026
f43f2f6

Attach notifier to broker host connection


wdcui Jul 1, 2026
8825534

Emit readiness notifications from event handling


wdcui Jul 1, 2026
f85dc25

Avoid duplicate readiness notifications


wdcui Jul 1, 2026
c37ef5a

Exercise readiness responses through serve_connection


wdcui Jul 1, 2026
8c1e075

Inline broker host request loop


wdcui Jul 1, 2026
3eb6ec5

Rename broker control socket argument


wdcui Jul 1, 2026
7f9b074

Clarify notification receiver thread ownership


wdcui Jul 1, 2026
9a185f3

Simplify broker socket option handling


wdcui Jul 1, 2026
5597b37

Require broker socket paths in connect


wdcui Jul 1, 2026
73f09bc

Merge broker socket retry helpers


wdcui Jul 1, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


152 changes: 121 additions & 31 deletions
litebox_broker_host/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -14,7 +14,9 @@ extern crate std;

use litebox_broker_core::{BrokerCore, BrokerSession, CallerCredential};
use litebox_broker_protocol::BROKER_PROTOCOL_VERSION;
use litebox_broker_protocol::channel::{HostControlChannel, HostReceive, PeerCredential};
use litebox_broker_protocol::channel::{
 HostControlChannel, HostNotificationChannel, HostReceive, PeerCredential,
};
use litebox_broker_protocol::error::ErrorCode;
use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse, WaitEventResponse};
use litebox_broker_protocol::message::{
Expand All@@ -25,15 +27,24 @@ mod error;

pub use error::{BrokerHostError, Result};

/// Authenticates, negotiates, and serves one broker connection over the control channel.
pub fn serve_connection<Channel>(
/// Authenticates, negotiates, and serves one broker association over paired
/// control and notification channels.
///
/// The deployment must bind both channels to the same authenticated peer
/// association. Active requests and responses remain on the control channel;
/// broker-initiated readiness wakeups are sent on the notification channel.
/// Event mutations caused by control requests return readiness in their control
/// response and do not also emit a duplicate notification.
pub fn serve_connection<ControlChannel, NotificationChannel, ChannelError>(
 core: &BrokerCore,
 channel: &mut Channel,
) -> Result<ConnectionTermination, Channel::Error>
 control_channel: &mut ControlChannel,
 _notification_channel: &mut NotificationChannel,
) -> Result<ConnectionTermination, ChannelError>
where
 Channel: HostControlChannel,
 ControlChannel: HostControlChannel<Error = ChannelError>,
 NotificationChannel: HostNotificationChannel<Error = ChannelError>,
{
 let peer_credential = channel
 let peer_credential = control_channel
 .peer_credential()
 .map_err(BrokerHostError::Channel)?;
 let caller_credential = match peer_credential {
Expand All@@ -43,13 +54,13 @@ where
 let session = core.create_session(caller_credential)?;

 loop {
 let request = match channel
 let request = match control_channel
 .recv_handshake_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_handshake_response(&BrokerHandshakeResponse::Error(
 ErrorCode::ProtocolState,
 ))
Expand All@@ -69,38 +80,31 @@ where
 broker_protocol_version: BROKER_PROTOCOL_VERSION,
 }
 };
channel
control_channel
 .send_handshake_response(&response)
 .map_err(BrokerHostError::Channel)?;
 if negotiated {
 break;
 }
 }

 serve_request_loop(channel, &session)
}

fn serve_request_loop<Channel>(
 channel: &mut Channel,
 session: &BrokerSession,
) -> Result<ConnectionTermination, Channel::Error>
where
 Channel: HostControlChannel,
{
 loop {
 let request = match channel.recv_request().map_err(BrokerHostError::Channel)? {
 let request = match control_channel
 .recv_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_response(&BrokerResponse::Error(ErrorCode::ProtocolState))
 .map_err(BrokerHostError::Channel)?;
 return Ok(ConnectionTermination::ProtocolViolation);
 }
 HostReceive::PeerClosed => break,
 };

 let response = handle_request(session, request);
channel
 let response = handle_request(&session, request);
control_channel
 .send_response(&response)
 .map_err(BrokerHostError::Channel)?;
 }
Expand DownExpand Up@@ -167,8 +171,11 @@ pub enum ConnectionTermination {
mod tests {
 use super::*;
 use litebox_broker_core::{PolicyEngine, PrincipalRights};
 use litebox_broker_protocol::event::{CreateEventRequest, WaitEventRequest};
 use litebox_broker_protocol::message::BrokerHandshakeRequest;
 use litebox_broker_protocol::event::{
 AddEventRequest, ConsumeEventRequest, CreateEventRequest, EventConsumeMode,
 WaitEventRequest,
 };
 use litebox_broker_protocol::message::{BrokerHandshakeRequest, BrokerNotification};
 use litebox_broker_protocol::{ObjectHandle, ProtocolVersion};

 #[test]
Expand All@@ -183,6 +190,7 @@ mod tests {
 serve_connection_rejects_active_request_before_negotiation(&broker);
 serve_connection_rejects_handshake_request_after_negotiation(&broker);
 serve_connection_returns_channel_error_when_response_send_fails(&broker);
 serve_connection_returns_event_readiness_in_control_responses(&broker);
 active_request_closes_object_reference(&broker);
 }

Expand All@@ -198,9 +206,10 @@ mod tests {
 Ok(HostReceive::PeerClosed),
 ]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand DownExpand Up@@ -228,9 +237,10 @@ mod tests {
 ]),
 std::vec::Vec::from([Ok(HostReceive::PeerClosed)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand All@@ -251,9 +261,10 @@ mod tests {
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 std::vec::Vec::new(),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -270,9 +281,10 @@ mod tests {
 }))]),
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -295,14 +307,54 @@ mod tests {
 std::vec::Vec::new(),
 );
 channel.send_error = true;
 let mut notifications = FakeHostNotificationChannel::default();

 match serve_connection(broker, &mut channel) {
 match serve_connection(broker, &mut channel, &mut notifications) {
 Err(BrokerHostError::Channel(())) => {}
 result => panic!("unexpected serve result: {result:?}"),
 }
 assert!(channel.handshake_responses.is_empty());
 }

 fn serve_connection_returns_event_readiness_in_control_responses(broker: &BrokerCore) {
 let mut channel = FakeHostControlChannel::new(
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerHandshakeRequest {
 protocol_version: BROKER_PROTOCOL_VERSION,
 }))]),
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Create(CreateEventRequest { initial_count: 0 }),
 )))]),
 );
 channel.enqueue_readiness_requests_after_create = true;
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert!(notifications.notifications.is_empty());
 assert_eq!(
 &channel.responses[1..],
 [
 BrokerResponse::Event(EventResponse::Add(AddEventResponse {
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: true,
 write_ready: true,
 },
 })),
 BrokerResponse::Event(EventResponse::Consume(
 litebox_broker_protocol::event::ConsumeEventResponse {
 value: 1,
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: false,
 write_ready: true,
 },
 }
 )),
 ]
 );
 }

 fn active_request_closes_object_reference(broker: &BrokerCore) {
 let session = broker
 .create_session(CallerCredential::Unauthenticated)
Expand DownExpand Up@@ -344,6 +396,7 @@ mod tests {
 requests: std::vec::Vec<core::result::Result<HostReceive<BrokerRequest>, ()>>,
 handshake_responses: std::vec::Vec<BrokerHandshakeResponse>,
 responses: std::vec::Vec<BrokerResponse>,
 enqueue_readiness_requests_after_create: bool,
 send_error: bool,
 }

Expand All@@ -359,6 +412,7 @@ mod tests {
 requests,
 handshake_responses: std::vec::Vec::new(),
 responses: std::vec::Vec::new(),
 enqueue_readiness_requests_after_create: false,
 send_error: false,
 }
 }
Expand DownExpand Up@@ -409,8 +463,44 @@ mod tests {
 if self.send_error {
 return Err(());
 }
 if self.enqueue_readiness_requests_after_create
 && let BrokerResponse::Event(EventResponse::Create(response)) = response
 {
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Add(AddEventRequest {
 handle: response.handle,
 value: 1,
 }),
 ))));
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Consume(ConsumeEventRequest {
 handle: response.handle,
 mode: EventConsumeMode::One,
 }),
 ))));
 self.requests.push(Ok(HostReceive::PeerClosed));
 }
 self.responses.push(response.clone());
 Ok(())
 }
 }

 #[derive(Default)]
 struct FakeHostNotificationChannel {
 notifications: std::vec::Vec<BrokerNotification>,
 }

 impl HostNotificationChannel for FakeHostNotificationChannel {
 type Error = ();

 fn send_notification(
 &mut self,
 notification: &BrokerNotification,
 ) -> core::result::Result<(), Self::Error> {
 self.notifications.push(notification.clone());
 Ok(())
 }
 }
}
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
Wire broker readiness notifications by wdcui · Pull Request #992 · microsoft/litebox · GitHub
Skip to content
Wire broker readiness notifications#992

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Weidong Cui (wdcui)
merged 15 commits into

from

Jul 2, 2026
 Merged


Wire broker readiness notifications
#992
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


15 commits


Select commit
Hold shift + click to select a range
b011ebf

Wire broker readiness notifications


wdcui Jul 1, 2026
1f2007e

Clarify broker host channel generic names


wdcui Jul 1, 2026
f4c14a2

Require broker notification channel


wdcui Jul 1, 2026
3b6cd95

Rename paired broker serving API


wdcui Jul 1, 2026
a6ae87d

Align broker host channel error bounds


wdcui Jul 1, 2026
f43f2f6

Attach notifier to broker host connection


wdcui Jul 1, 2026
8825534

Emit readiness notifications from event handling


wdcui Jul 1, 2026
f85dc25

Avoid duplicate readiness notifications


wdcui Jul 1, 2026
c37ef5a

Exercise readiness responses through serve_connection


wdcui Jul 1, 2026
8c1e075

Inline broker host request loop


wdcui Jul 1, 2026
3eb6ec5

Rename broker control socket argument


wdcui Jul 1, 2026
7f9b074

Clarify notification receiver thread ownership


wdcui Jul 1, 2026
9a185f3

Simplify broker socket option handling


wdcui Jul 1, 2026
5597b37

Require broker socket paths in connect


wdcui Jul 1, 2026
73f09bc

Merge broker socket retry helpers


wdcui Jul 1, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


152 changes: 121 additions & 31 deletions
litebox_broker_host/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -14,7 +14,9 @@ extern crate std;

use litebox_broker_core::{BrokerCore, BrokerSession, CallerCredential};
use litebox_broker_protocol::BROKER_PROTOCOL_VERSION;
use litebox_broker_protocol::channel::{HostControlChannel, HostReceive, PeerCredential};
use litebox_broker_protocol::channel::{
 HostControlChannel, HostNotificationChannel, HostReceive, PeerCredential,
};
use litebox_broker_protocol::error::ErrorCode;
use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse, WaitEventResponse};
use litebox_broker_protocol::message::{
Expand All@@ -25,15 +27,24 @@ mod error;

pub use error::{BrokerHostError, Result};

/// Authenticates, negotiates, and serves one broker connection over the control channel.
pub fn serve_connection<Channel>(
/// Authenticates, negotiates, and serves one broker association over paired
/// control and notification channels.
///
/// The deployment must bind both channels to the same authenticated peer
/// association. Active requests and responses remain on the control channel;
/// broker-initiated readiness wakeups are sent on the notification channel.
/// Event mutations caused by control requests return readiness in their control
/// response and do not also emit a duplicate notification.
pub fn serve_connection<ControlChannel, NotificationChannel, ChannelError>(
 core: &BrokerCore,
 channel: &mut Channel,
) -> Result<ConnectionTermination, Channel::Error>
 control_channel: &mut ControlChannel,
 _notification_channel: &mut NotificationChannel,
) -> Result<ConnectionTermination, ChannelError>
where
 Channel: HostControlChannel,
 ControlChannel: HostControlChannel<Error = ChannelError>,
 NotificationChannel: HostNotificationChannel<Error = ChannelError>,
{
 let peer_credential = channel
 let peer_credential = control_channel
 .peer_credential()
 .map_err(BrokerHostError::Channel)?;
 let caller_credential = match peer_credential {
Expand All@@ -43,13 +54,13 @@ where
 let session = core.create_session(caller_credential)?;

 loop {
 let request = match channel
 let request = match control_channel
 .recv_handshake_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_handshake_response(&BrokerHandshakeResponse::Error(
 ErrorCode::ProtocolState,
 ))
Expand All@@ -69,38 +80,31 @@ where
 broker_protocol_version: BROKER_PROTOCOL_VERSION,
 }
 };
channel
control_channel
 .send_handshake_response(&response)
 .map_err(BrokerHostError::Channel)?;
 if negotiated {
 break;
 }
 }

 serve_request_loop(channel, &session)
}

fn serve_request_loop<Channel>(
 channel: &mut Channel,
 session: &BrokerSession,
) -> Result<ConnectionTermination, Channel::Error>
where
 Channel: HostControlChannel,
{
 loop {
 let request = match channel.recv_request().map_err(BrokerHostError::Channel)? {
 let request = match control_channel
 .recv_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_response(&BrokerResponse::Error(ErrorCode::ProtocolState))
 .map_err(BrokerHostError::Channel)?;
 return Ok(ConnectionTermination::ProtocolViolation);
 }
 HostReceive::PeerClosed => break,
 };

 let response = handle_request(session, request);
channel
 let response = handle_request(&session, request);
control_channel
 .send_response(&response)
 .map_err(BrokerHostError::Channel)?;
 }
Expand DownExpand Up@@ -167,8 +171,11 @@ pub enum ConnectionTermination {
mod tests {
 use super::*;
 use litebox_broker_core::{PolicyEngine, PrincipalRights};
 use litebox_broker_protocol::event::{CreateEventRequest, WaitEventRequest};
 use litebox_broker_protocol::message::BrokerHandshakeRequest;
 use litebox_broker_protocol::event::{
 AddEventRequest, ConsumeEventRequest, CreateEventRequest, EventConsumeMode,
 WaitEventRequest,
 };
 use litebox_broker_protocol::message::{BrokerHandshakeRequest, BrokerNotification};
 use litebox_broker_protocol::{ObjectHandle, ProtocolVersion};

 #[test]
Expand All@@ -183,6 +190,7 @@ mod tests {
 serve_connection_rejects_active_request_before_negotiation(&broker);
 serve_connection_rejects_handshake_request_after_negotiation(&broker);
 serve_connection_returns_channel_error_when_response_send_fails(&broker);
 serve_connection_returns_event_readiness_in_control_responses(&broker);
 active_request_closes_object_reference(&broker);
 }

Expand All@@ -198,9 +206,10 @@ mod tests {
 Ok(HostReceive::PeerClosed),
 ]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand DownExpand Up@@ -228,9 +237,10 @@ mod tests {
 ]),
 std::vec::Vec::from([Ok(HostReceive::PeerClosed)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand All@@ -251,9 +261,10 @@ mod tests {
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 std::vec::Vec::new(),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -270,9 +281,10 @@ mod tests {
 }))]),
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -295,14 +307,54 @@ mod tests {
 std::vec::Vec::new(),
 );
 channel.send_error = true;
 let mut notifications = FakeHostNotificationChannel::default();

 match serve_connection(broker, &mut channel) {
 match serve_connection(broker, &mut channel, &mut notifications) {
 Err(BrokerHostError::Channel(())) => {}
 result => panic!("unexpected serve result: {result:?}"),
 }
 assert!(channel.handshake_responses.is_empty());
 }

 fn serve_connection_returns_event_readiness_in_control_responses(broker: &BrokerCore) {
 let mut channel = FakeHostControlChannel::new(
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerHandshakeRequest {
 protocol_version: BROKER_PROTOCOL_VERSION,
 }))]),
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Create(CreateEventRequest { initial_count: 0 }),
 )))]),
 );
 channel.enqueue_readiness_requests_after_create = true;
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert!(notifications.notifications.is_empty());
 assert_eq!(
 &channel.responses[1..],
 [
 BrokerResponse::Event(EventResponse::Add(AddEventResponse {
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: true,
 write_ready: true,
 },
 })),
 BrokerResponse::Event(EventResponse::Consume(
 litebox_broker_protocol::event::ConsumeEventResponse {
 value: 1,
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: false,
 write_ready: true,
 },
 }
 )),
 ]
 );
 }

 fn active_request_closes_object_reference(broker: &BrokerCore) {
 let session = broker
 .create_session(CallerCredential::Unauthenticated)
Expand DownExpand Up@@ -344,6 +396,7 @@ mod tests {
 requests: std::vec::Vec<core::result::Result<HostReceive<BrokerRequest>, ()>>,
 handshake_responses: std::vec::Vec<BrokerHandshakeResponse>,
 responses: std::vec::Vec<BrokerResponse>,
 enqueue_readiness_requests_after_create: bool,
 send_error: bool,
 }

Expand All@@ -359,6 +412,7 @@ mod tests {
 requests,
 handshake_responses: std::vec::Vec::new(),
 responses: std::vec::Vec::new(),
 enqueue_readiness_requests_after_create: false,
 send_error: false,
 }
 }
Expand DownExpand Up@@ -409,8 +463,44 @@ mod tests {
 if self.send_error {
 return Err(());
 }
 if self.enqueue_readiness_requests_after_create
 && let BrokerResponse::Event(EventResponse::Create(response)) = response
 {
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Add(AddEventRequest {
 handle: response.handle,
 value: 1,
 }),
 ))));
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Consume(ConsumeEventRequest {
 handle: response.handle,
 mode: EventConsumeMode::One,
 }),
 ))));
 self.requests.push(Ok(HostReceive::PeerClosed));
 }
 self.responses.push(response.clone());
 Ok(())
 }
 }

 #[derive(Default)]
 struct FakeHostNotificationChannel {
 notifications: std::vec::Vec<BrokerNotification>,
 }

 impl HostNotificationChannel for FakeHostNotificationChannel {
 type Error = ();

 fn send_notification(
 &mut self,
 notification: &BrokerNotification,
 ) -> core::result::Result<(), Self::Error> {
 self.notifications.push(notification.clone());
 Ok(())
 }
 }
}
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();
Wire broker readiness notifications by wdcui · Pull Request #992 · microsoft/litebox · GitHub
Skip to content
Wire broker readiness notifications#992

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Merged

Weidong Cui (wdcui)
merged 15 commits into

from

Jul 2, 2026
 Merged


Wire broker readiness notifications
#992
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


15 commits


Select commit
Hold shift + click to select a range
b011ebf

Wire broker readiness notifications


wdcui Jul 1, 2026
1f2007e

Clarify broker host channel generic names


wdcui Jul 1, 2026
f4c14a2

Require broker notification channel


wdcui Jul 1, 2026
3b6cd95

Rename paired broker serving API


wdcui Jul 1, 2026
a6ae87d

Align broker host channel error bounds


wdcui Jul 1, 2026
f43f2f6

Attach notifier to broker host connection


wdcui Jul 1, 2026
8825534

Emit readiness notifications from event handling


wdcui Jul 1, 2026
f85dc25

Avoid duplicate readiness notifications


wdcui Jul 1, 2026
c37ef5a

Exercise readiness responses through serve_connection


wdcui Jul 1, 2026
8c1e075

Inline broker host request loop


wdcui Jul 1, 2026
3eb6ec5

Rename broker control socket argument


wdcui Jul 1, 2026
7f9b074

Clarify notification receiver thread ownership


wdcui Jul 1, 2026
9a185f3

Simplify broker socket option handling


wdcui Jul 1, 2026
5597b37

Require broker socket paths in connect


wdcui Jul 1, 2026
73f09bc

Merge broker socket retry helpers


wdcui Jul 1, 2026

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


152 changes: 121 additions & 31 deletions
litebox_broker_host/src/lib.rs

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -14,7 +14,9 @@ extern crate std;

use litebox_broker_core::{BrokerCore, BrokerSession, CallerCredential};
use litebox_broker_protocol::BROKER_PROTOCOL_VERSION;
use litebox_broker_protocol::channel::{HostControlChannel, HostReceive, PeerCredential};
use litebox_broker_protocol::channel::{
 HostControlChannel, HostNotificationChannel, HostReceive, PeerCredential,
};
use litebox_broker_protocol::error::ErrorCode;
use litebox_broker_protocol::event::{AddEventResponse, CreateEventResponse, WaitEventResponse};
use litebox_broker_protocol::message::{
Expand All@@ -25,15 +27,24 @@ mod error;

pub use error::{BrokerHostError, Result};

/// Authenticates, negotiates, and serves one broker connection over the control channel.
pub fn serve_connection<Channel>(
/// Authenticates, negotiates, and serves one broker association over paired
/// control and notification channels.
///
/// The deployment must bind both channels to the same authenticated peer
/// association. Active requests and responses remain on the control channel;
/// broker-initiated readiness wakeups are sent on the notification channel.
/// Event mutations caused by control requests return readiness in their control
/// response and do not also emit a duplicate notification.
pub fn serve_connection<ControlChannel, NotificationChannel, ChannelError>(
 core: &BrokerCore,
 channel: &mut Channel,
) -> Result<ConnectionTermination, Channel::Error>
 control_channel: &mut ControlChannel,
 _notification_channel: &mut NotificationChannel,
) -> Result<ConnectionTermination, ChannelError>
where
 Channel: HostControlChannel,
 ControlChannel: HostControlChannel<Error = ChannelError>,
 NotificationChannel: HostNotificationChannel<Error = ChannelError>,
{
 let peer_credential = channel
 let peer_credential = control_channel
 .peer_credential()
 .map_err(BrokerHostError::Channel)?;
 let caller_credential = match peer_credential {
Expand All@@ -43,13 +54,13 @@ where
 let session = core.create_session(caller_credential)?;

 loop {
 let request = match channel
 let request = match control_channel
 .recv_handshake_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_handshake_response(&BrokerHandshakeResponse::Error(
 ErrorCode::ProtocolState,
 ))
Expand All@@ -69,38 +80,31 @@ where
 broker_protocol_version: BROKER_PROTOCOL_VERSION,
 }
 };
channel
control_channel
 .send_handshake_response(&response)
 .map_err(BrokerHostError::Channel)?;
 if negotiated {
 break;
 }
 }

 serve_request_loop(channel, &session)
}

fn serve_request_loop<Channel>(
 channel: &mut Channel,
 session: &BrokerSession,
) -> Result<ConnectionTermination, Channel::Error>
where
 Channel: HostControlChannel,
{
 loop {
 let request = match channel.recv_request().map_err(BrokerHostError::Channel)? {
 let request = match control_channel
 .recv_request()
 .map_err(BrokerHostError::Channel)?
 {
 HostReceive::Message(request) => request,
 HostReceive::ProtocolViolation => {
channel
control_channel
 .send_response(&BrokerResponse::Error(ErrorCode::ProtocolState))
 .map_err(BrokerHostError::Channel)?;
 return Ok(ConnectionTermination::ProtocolViolation);
 }
 HostReceive::PeerClosed => break,
 };

 let response = handle_request(session, request);
channel
 let response = handle_request(&session, request);
control_channel
 .send_response(&response)
 .map_err(BrokerHostError::Channel)?;
 }
Expand DownExpand Up@@ -167,8 +171,11 @@ pub enum ConnectionTermination {
mod tests {
 use super::*;
 use litebox_broker_core::{PolicyEngine, PrincipalRights};
 use litebox_broker_protocol::event::{CreateEventRequest, WaitEventRequest};
 use litebox_broker_protocol::message::BrokerHandshakeRequest;
 use litebox_broker_protocol::event::{
 AddEventRequest, ConsumeEventRequest, CreateEventRequest, EventConsumeMode,
 WaitEventRequest,
 };
 use litebox_broker_protocol::message::{BrokerHandshakeRequest, BrokerNotification};
 use litebox_broker_protocol::{ObjectHandle, ProtocolVersion};

 #[test]
Expand All@@ -183,6 +190,7 @@ mod tests {
 serve_connection_rejects_active_request_before_negotiation(&broker);
 serve_connection_rejects_handshake_request_after_negotiation(&broker);
 serve_connection_returns_channel_error_when_response_send_fails(&broker);
 serve_connection_returns_event_readiness_in_control_responses(&broker);
 active_request_closes_object_reference(&broker);
 }

Expand All@@ -198,9 +206,10 @@ mod tests {
 Ok(HostReceive::PeerClosed),
 ]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand DownExpand Up@@ -228,9 +237,10 @@ mod tests {
 ]),
 std::vec::Vec::from([Ok(HostReceive::PeerClosed)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert_eq!(
Expand All@@ -251,9 +261,10 @@ mod tests {
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 std::vec::Vec::new(),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -270,9 +281,10 @@ mod tests {
 }))]),
 std::vec::Vec::from([Ok(HostReceive::ProtocolViolation)]),
 );
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel).unwrap(),
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::ProtocolViolation
 );
 assert_eq!(
Expand All@@ -295,14 +307,54 @@ mod tests {
 std::vec::Vec::new(),
 );
 channel.send_error = true;
 let mut notifications = FakeHostNotificationChannel::default();

 match serve_connection(broker, &mut channel) {
 match serve_connection(broker, &mut channel, &mut notifications) {
 Err(BrokerHostError::Channel(())) => {}
 result => panic!("unexpected serve result: {result:?}"),
 }
 assert!(channel.handshake_responses.is_empty());
 }

 fn serve_connection_returns_event_readiness_in_control_responses(broker: &BrokerCore) {
 let mut channel = FakeHostControlChannel::new(
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerHandshakeRequest {
 protocol_version: BROKER_PROTOCOL_VERSION,
 }))]),
 std::vec::Vec::from([Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Create(CreateEventRequest { initial_count: 0 }),
 )))]),
 );
 channel.enqueue_readiness_requests_after_create = true;
 let mut notifications = FakeHostNotificationChannel::default();

 assert_eq!(
 serve_connection(broker, &mut channel, &mut notifications).unwrap(),
 ConnectionTermination::PeerClosed
 );
 assert!(notifications.notifications.is_empty());
 assert_eq!(
 &channel.responses[1..],
 [
 BrokerResponse::Event(EventResponse::Add(AddEventResponse {
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: true,
 write_ready: true,
 },
 })),
 BrokerResponse::Event(EventResponse::Consume(
 litebox_broker_protocol::event::ConsumeEventResponse {
 value: 1,
 readiness: litebox_broker_protocol::event::ReadinessState {
 read_ready: false,
 write_ready: true,
 },
 }
 )),
 ]
 );
 }

 fn active_request_closes_object_reference(broker: &BrokerCore) {
 let session = broker
 .create_session(CallerCredential::Unauthenticated)
Expand DownExpand Up@@ -344,6 +396,7 @@ mod tests {
 requests: std::vec::Vec<core::result::Result<HostReceive<BrokerRequest>, ()>>,
 handshake_responses: std::vec::Vec<BrokerHandshakeResponse>,
 responses: std::vec::Vec<BrokerResponse>,
 enqueue_readiness_requests_after_create: bool,
 send_error: bool,
 }

Expand All@@ -359,6 +412,7 @@ mod tests {
 requests,
 handshake_responses: std::vec::Vec::new(),
 responses: std::vec::Vec::new(),
 enqueue_readiness_requests_after_create: false,
 send_error: false,
 }
 }
Expand DownExpand Up@@ -409,8 +463,44 @@ mod tests {
 if self.send_error {
 return Err(());
 }
 if self.enqueue_readiness_requests_after_create
 && let BrokerResponse::Event(EventResponse::Create(response)) = response
 {
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Add(AddEventRequest {
 handle: response.handle,
 value: 1,
 }),
 ))));
 self.requests
 .push(Ok(HostReceive::Message(BrokerRequest::Event(
 EventRequest::Consume(ConsumeEventRequest {
 handle: response.handle,
 mode: EventConsumeMode::One,
 }),
 ))));
 self.requests.push(Ok(HostReceive::PeerClosed));
 }
 self.responses.push(response.clone());
 Ok(())
 }
 }

 #[derive(Default)]
 struct FakeHostNotificationChannel {
 notifications: std::vec::Vec<BrokerNotification>,
 }

 impl HostNotificationChannel for FakeHostNotificationChannel {
 type Error = ();

 fn send_notification(
 &mut self,
 notification: &BrokerNotification,
 ) -> core::result::Result<(), Self::Error> {
 self.notifications.push(notification.clone());
 Ok(())
 }
 }
}
Loading