diff --git a/lib/ble/ble_winrt.cpp b/lib/ble/ble_winrt.cpp index 8e4cad40..3656039d 100644 --- a/lib/ble/ble_winrt.cpp +++ b/lib/ble/ble_winrt.cpp @@ -63,12 +63,12 @@ struct WinLink { bool outgoing = false, opened = false, sending = false; BluetoothLEDevice device{nullptr}; GattDeviceService service{nullptr}; - GattCharacteristic rx{nullptr}, tx{nullptr}, invite{nullptr}; + GattCharacteristic rx{nullptr}, tx{nullptr}, invite{nullptr}, ack{nullptr}; GattSession session{nullptr}; BluetoothLEPreferredConnectionParametersRequest speed_request{nullptr}; /* Живёт до закрытия link. */ - GattSubscribedClient subscriber{nullptr}, invite_subscriber{nullptr}; + GattSubscribedClient subscriber{nullptr}, invite_subscriber{nullptr}, ack_subscriber{nullptr}; IAsyncInfo operation{nullptr}; - GattCharacteristic::ValueChanged_revoker value_changed, invite_changed; + GattCharacteristic::ValueChanged_revoker value_changed, invite_changed, ack_changed; GattSession::SessionStatusChanged_revoker session_changed; GattSession::MaxPduSizeChanged_revoker pdu_changed; BluetoothLEDevice::ConnectionParametersChanged_revoker parameters_changed; @@ -91,13 +91,13 @@ struct WinState : std::enable_shared_from_this { BluetoothLEAdvertisementWatcher watcher{nullptr}; BluetoothLEAdvertisementPublisher publisher{nullptr}; GattServiceProvider provider{nullptr}; - GattLocalCharacteristic local_rx{nullptr}, local_tx{nullptr}, local_invite{nullptr}; + GattLocalCharacteristic local_rx{nullptr}, local_tx{nullptr}, local_invite{nullptr}, local_ack{nullptr}; BluetoothLEAdvertisementWatcher::Received_revoker received; BluetoothLEAdvertisementWatcher::Stopped_revoker scan_stopped; BluetoothLEAdvertisementPublisher::StatusChanged_revoker publish_changed; GattServiceProvider::AdvertisementStatusChanged_revoker service_changed; - GattLocalCharacteristic::WriteRequested_revoker write_requested, invite_write_requested; - GattLocalCharacteristic::SubscribedClientsChanged_revoker subscriptions_changed, invite_subscriptions_changed; + GattLocalCharacteristic::WriteRequested_revoker write_requested, invite_write_requested, ack_write_requested; + GattLocalCharacteristic::SubscribedClientsChanged_revoker subscriptions_changed, invite_subscriptions_changed, ack_subscriptions_changed; std::map> links; std::array broadcast{}; size_t broadcast_size = 0; @@ -245,16 +245,18 @@ struct WinState : std::enable_shared_from_this { } void subscriptions() { std::vector active; + auto invitations = local_invite.SubscribedClients(), acknowledgements = local_ack.SubscribedClients(); for (auto const& client : local_tx.SubscribedClients()) { - for (auto const& invitation : local_invite.SubscribedClients()) { - if (client.Session().DeviceId().Id() != invitation.Session().DeviceId().Id()) continue; - auto* l = incoming(client.Session()); if (!l) continue; - l->subscriber = client; l->invite_subscriber = invitation; active.push_back(l->id); - if (!l->opened) { - auto capacity = std::min(client.MaxNotificationSize(), invitation.MaxNotificationSize()); - DEBUG_DEBUG(CAT, "BLE WinRT subscribed both channels link=%llu capacity=%u", (unsigned long long)l->id, capacity); - l->opened = true; ble_backend_opened(owner, l->id, capacity + 3); - } + auto same_device = [&client](auto const& other) { return client.Session().DeviceId().Id() == other.Session().DeviceId().Id(); }; + auto invitation = std::find_if(invitations.begin(), invitations.end(), same_device); + auto acknowledgement = std::find_if(acknowledgements.begin(), acknowledgements.end(), same_device); + if (invitation == invitations.end() || acknowledgement == acknowledgements.end()) continue; + auto* l = incoming(client.Session()); if (!l) continue; + l->subscriber = client; l->invite_subscriber = *invitation; l->ack_subscriber = *acknowledgement; active.push_back(l->id); + if (!l->opened) { + auto capacity = std::min({client.MaxNotificationSize(), invitation->MaxNotificationSize(), acknowledgement->MaxNotificationSize()}); + DEBUG_DEBUG(CAT, "BLE WinRT subscribed DATA/INVITE/ACK link=%llu capacity=%u", (unsigned long long)l->id, capacity); + l->opened = true; ble_backend_opened(owner, l->id, capacity + 3); } } std::vector gone; @@ -291,10 +293,20 @@ struct WinState : std::enable_shared_from_this { await("create invitation characteristic", provider.Service().CreateCharacteristicAsync(guid(BLE_INVITE_UUID), parameters), 0, [](WinState& s, GattLocalCharacteristicResult const& result) { if (result.Error() != BluetoothError::Success) { s.radio_error("create invitation characteristic", result.Error()); return; } - s.local_invite = result.Characteristic(); s.start_server(); + s.local_invite = result.Characteristic(); s.make_ack(); }); } - GattLocalCharacteristic::WriteRequested_revoker watch_write(GattLocalCharacteristic const& characteristic, enum ble_channel channel) { + void make_ack() { + GattLocalCharacteristicParameters parameters; + parameters.CharacteristicProperties(GattCharacteristicProperties::Write | GattCharacteristicProperties::Indicate); + parameters.WriteProtectionLevel(GattProtectionLevel::Plain); parameters.ReadProtectionLevel(GattProtectionLevel::Plain); + await("create ACK characteristic", provider.Service().CreateCharacteristicAsync(guid(BLE_ACK_UUID), parameters), 0, + [](WinState& s, GattLocalCharacteristicResult const& result) { + if (result.Error() != BluetoothError::Success) { s.radio_error("create ACK characteristic", result.Error()); return; } + s.local_ack = result.Characteristic(); s.start_server(); + }); + } + GattLocalCharacteristic::WriteRequested_revoker watch_write(GattLocalCharacteristic const& characteristic, enum ble_wire_channel channel) { auto weak = weak_from_this(); return characteristic.WriteRequested(auto_revoke, [weak, channel](auto const&, GattWriteRequestedEventArgs const& args) { try { @@ -308,7 +320,7 @@ struct WinState : std::enable_shared_from_this { } auto request = operation.GetResults(); if (!request) { DEBUG_WARN(CAT, "BLE WinRT expired write request"); return; } bool response = request.Option() == GattWriteOption::WriteWithResponse; - auto expected = channel == BLE_CHANNEL_DATA ? GattWriteOption::WriteWithoutResponse : GattWriteOption::WriteWithResponse; + auto expected = channel == BLE_WIRE_DATA ? GattWriteOption::WriteWithoutResponse : GattWriteOption::WriteWithResponse; if (request.Offset() || request.Option() != expected || !request.Value().Length() || request.Value().Length() > BLE_VALUE_MAX) { DEBUG_WARN(CAT, "BLE WinRT invalid write offset=%u option=%d bytes=%u", request.Offset(), (int)request.Option(), request.Value().Length()); @@ -331,8 +343,12 @@ struct WinState : std::enable_shared_from_this { } void start_server() { auto weak = weak_from_this(); - write_requested = watch_write(local_rx, BLE_CHANNEL_DATA); - invite_write_requested = watch_write(local_invite, BLE_CHANNEL_INVITE); + write_requested = watch_write(local_rx, BLE_WIRE_DATA); + invite_write_requested = watch_write(local_invite, BLE_WIRE_INVITE); + ack_write_requested = watch_write(local_ack, BLE_WIRE_ACK); + ack_subscriptions_changed = local_ack.SubscribedClientsChanged(auto_revoke, [weak](auto const&, auto const&) { + if (auto s = weak.lock()) s->post([](WinState& s) { s.subscriptions(); }); + }); invite_subscriptions_changed = local_invite.SubscribedClientsChanged(auto_revoke, [weak](auto const&, auto const&) { if (auto s = weak.lock()) s->post([](WinState& s) { s.subscriptions(); }); }); @@ -447,8 +463,9 @@ struct WinState : std::enable_shared_from_this { if (ch.Uuid() == guid(BLE_RX_UUID)) l->rx = ch; if (ch.Uuid() == guid(BLE_TX_UUID)) l->tx = ch; if (ch.Uuid() == guid(BLE_INVITE_UUID)) l->invite = ch; + if (ch.Uuid() == guid(BLE_ACK_UUID)) l->ack = ch; } - if (!l->rx || !l->tx || !l->invite) { s.error(id, "RX/TX/invitation characteristics missing"); return; } + if (!l->rx || !l->tx || !l->invite || !l->ack) { s.error(id, "RX/TX/INVITE/ACK characteristics missing"); return; } s.subscribe(id); }); }); @@ -463,12 +480,13 @@ struct WinState : std::enable_shared_from_this { s.prefer_speed(*l); DEBUG_DEBUG(CAT, "BLE WinRT session opened link=%llu mtu=%u maintain_connection=1", (unsigned long long)id, session.MaxPduSize()); s.log_connection(id); - l->value_changed = s.watch_value(id, l->tx, BLE_CHANNEL_DATA); - l->invite_changed = s.watch_value(id, l->invite, BLE_CHANNEL_INVITE); - s.subscribe_channel(id, BLE_CHANNEL_DATA); + l->value_changed = s.watch_value(id, l->tx, BLE_WIRE_DATA); + l->invite_changed = s.watch_value(id, l->invite, BLE_WIRE_INVITE); + l->ack_changed = s.watch_value(id, l->ack, BLE_WIRE_ACK); + s.subscribe_channel(id, BLE_WIRE_DATA); }); } - GattCharacteristic::ValueChanged_revoker watch_value(ble_link_id id, GattCharacteristic const& characteristic, enum ble_channel channel) { + GattCharacteristic::ValueChanged_revoker watch_value(ble_link_id id, GattCharacteristic const& characteristic, enum ble_wire_channel channel) { auto weak = weak_from_this(); return characteristic.ValueChanged(auto_revoke, [weak, id, channel](auto const&, GattValueChangedEventArgs const& args) { if (auto s = weak.lock()) s->post([id, channel, args](WinState& s) { @@ -478,10 +496,10 @@ struct WinState : std::enable_shared_from_this { }); }); } - void subscribe_channel(ble_link_id id, enum ble_channel channel) { + void subscribe_channel(ble_link_id id, enum ble_wire_channel channel) { auto* l = link(id); - auto characteristic = channel == BLE_CHANNEL_DATA ? l->tx : l->invite; - auto mode = channel == BLE_CHANNEL_DATA ? GattClientCharacteristicConfigurationDescriptorValue::Notify : + auto characteristic = channel == BLE_WIRE_ACK ? l->ack : channel == BLE_WIRE_DATA ? l->tx : l->invite; + auto mode = channel == BLE_WIRE_DATA ? GattClientCharacteristicConfigurationDescriptorValue::Notify : GattClientCharacteristicConfigurationDescriptorValue::Indicate; await("subscribe channel", characteristic.WriteClientCharacteristicConfigurationDescriptorAsync(mode), id, [id, channel](WinState& s, GattCommunicationStatus status) { @@ -491,7 +509,8 @@ struct WinState : std::enable_shared_from_this { (unsigned long long)id, channel, (int)status); s.error(id, "GATT subscription failed"); return; } - if (channel == BLE_CHANNEL_DATA) { s.subscribe_channel(id, BLE_CHANNEL_INVITE); return; } + if (channel == BLE_WIRE_DATA) { s.subscribe_channel(id, BLE_WIRE_INVITE); return; } + if (channel == BLE_WIRE_INVITE) { s.subscribe_channel(id, BLE_WIRE_ACK); return; } auto* l = s.link(id); l->opened = true; ble_backend_opened(s.owner, id, l->session.MaxPduSize()); }); } @@ -500,7 +519,7 @@ struct WinState : std::enable_shared_from_this { auto l = std::move(it->second); links.erase(it); DEBUG_DEBUG(CAT, "BLE WinRT close link=%llu outgoing=%d opened=%d sending=%d pending=%d", (unsigned long long)id, l->outgoing, l->opened, l->sending, bool(l->operation)); - l->value_changed.revoke(); l->invite_changed.revoke(); l->session_changed.revoke(); l->pdu_changed.revoke(); + l->value_changed.revoke(); l->invite_changed.revoke(); l->ack_changed.revoke(); l->session_changed.revoke(); l->pdu_changed.revoke(); l->parameters_changed.revoke(); l->phy_changed.revoke(); try { if (l->speed_request) l->speed_request.Close(); } catch (...) { DEBUG_WARN(CAT, "%s", exception_reason("close throughput request", id).c_str()); } @@ -522,12 +541,14 @@ struct WinState : std::enable_shared_from_this { if (task) uasync_cancel_post(ua, task); received.revoke(); scan_stopped.revoke(); publish_changed.revoke(); service_changed.revoke(); write_requested.revoke(); subscriptions_changed.revoke(); invite_write_requested.revoke(); invite_subscriptions_changed.revoke(); + ack_write_requested.revoke(); ack_subscriptions_changed.revoke(); while (!links.empty()) close(links.begin()->first); try { if (startup) startup.Cancel(); } catch (...) { DEBUG_WARN(CAT, "%s", exception_reason("cancel startup").c_str()); } try { if (watcher) watcher.Stop(); } catch (...) { DEBUG_WARN(CAT, "%s", exception_reason("stop scan").c_str()); } try { if (publisher) publisher.Stop(); } catch (...) { DEBUG_WARN(CAT, "%s", exception_reason("stop publisher").c_str()); } try { if (provider) provider.StopAdvertising(); } catch (...) { DEBUG_WARN(CAT, "%s", exception_reason("stop server").c_str()); } - watcher = nullptr; publisher = nullptr; provider = nullptr; local_rx = nullptr; local_tx = nullptr; local_invite = nullptr; startup = nullptr; + watcher = nullptr; publisher = nullptr; provider = nullptr; + local_rx = nullptr; local_tx = nullptr; local_invite = nullptr; local_ack = nullptr; startup = nullptr; if (apartment) { uninit_apartment(); apartment = false; } } }; @@ -583,9 +604,9 @@ extern "C" int ble_backend_connect(struct ble_backend* b, ble_link_id id, const } catch (...) { DEBUG_ERROR(CAT, "%s", exception_reason("connect", id).c_str()); b->state->close(id); return BLE_ERROR; } } extern "C" void ble_backend_close(struct ble_backend* b, ble_link_id id) { b->state->close(id); } -extern "C" int ble_backend_send(struct ble_backend* b, ble_link_id id, enum ble_channel channel, const uint8_t* data, size_t len) { +extern "C" int ble_backend_send(struct ble_backend* b, ble_link_id id, enum ble_wire_channel channel, const uint8_t* data, size_t len) { auto& s = *b->state; auto* l = s.link(id); - if ((unsigned)channel >= BLE_CHANNEL_COUNT || !l || !l->opened || l->sending) { + if ((unsigned)channel >= BLE_WIRE_COUNT || !l || !l->opened || l->sending) { DEBUG_ERROR(CAT, "BLE invalid WinRT send link=%llu exists=%d opened=%d sending=%d bytes=%zu", (unsigned long long)id, l != nullptr, l && l->opened, l && l->sending, len); return BLE_ERROR; @@ -593,8 +614,9 @@ extern "C" int ble_backend_send(struct ble_backend* b, ble_link_id id, enum ble_ try { DEBUG_TRACE(CAT, "BLE WinRT TX fragment link=%llu outgoing=%d bytes=%zu", (unsigned long long)id, l->outgoing, len); l->sending = true; auto buffer = WinState::buffer(data, len); - auto option = channel == BLE_CHANNEL_DATA ? GattWriteOption::WriteWithoutResponse : GattWriteOption::WriteWithResponse; - if (l->outgoing) s.await("write frame", (channel == BLE_CHANNEL_INVITE ? l->invite : l->rx).WriteValueWithResultAsync(buffer, option), id, + auto option = channel == BLE_WIRE_DATA ? GattWriteOption::WriteWithoutResponse : GattWriteOption::WriteWithResponse; + if (l->outgoing) s.await("write frame", (channel == BLE_WIRE_ACK ? l->ack : channel == BLE_WIRE_INVITE ? l->invite : l->rx) + .WriteValueWithResultAsync(buffer, option), id, [id, len](WinState& s, GattWriteResult const& result) { auto* l = s.link(id); l->sending = false; auto protocol = result.ProtocolError(); @@ -604,8 +626,10 @@ extern "C" int ble_backend_send(struct ble_backend* b, ble_link_id id, enum ble_ else DEBUG_TRACE(CAT, "BLE WinRT write complete link=%llu bytes=%zu", (unsigned long long)id, len); ble_backend_sent(s.owner, id, result.Status() == GattCommunicationStatus::Success); }); - else s.await("notify frame", (channel == BLE_CHANNEL_INVITE ? s.local_invite : s.local_tx).NotifyValueAsync(buffer, - channel == BLE_CHANNEL_INVITE ? l->invite_subscriber : l->subscriber), id, [id, len](WinState& s, GattClientNotificationResult const& result) { + else s.await("notify frame", (channel == BLE_WIRE_ACK ? s.local_ack : channel == BLE_WIRE_INVITE ? s.local_invite : s.local_tx) + .NotifyValueAsync(buffer, channel == BLE_WIRE_ACK ? l->ack_subscriber : + channel == BLE_WIRE_INVITE ? l->invite_subscriber : l->subscriber), id, + [id, len](WinState& s, GattClientNotificationResult const& result) { auto* l = s.link(id); l->sending = false; if (result.Status() != GattCommunicationStatus::Success || result.BytesSent() != len) DEBUG_WARN(CAT, "BLE WinRT notification failed link=%llu expected=%zu sent=%u status=%d", (unsigned long long)id, diff --git a/lib/ble/tests/test_ble_transport.c b/lib/ble/tests/test_ble_transport.c index 5705c528..fb25e9ac 100644 --- a/lib/ble/tests/test_ble_transport.c +++ b/lib/ble/tests/test_ble_transport.c @@ -110,10 +110,11 @@ static void exercise(unsigned int test_mtu) { e = ble_channel_rx_get(t[1], b, BLE_CHANNEL_INVITE); CHECK(e && e->len == 20); release(e); drop = 2; CHECK(!ble_tx_put(t[0], a, frame(1500, 0x44))); drain(ua); CHECK(!ble_rx_get(t[1], b)); CHECK(!ble_tx_put(t[0], a, frame(5, 0x55))); drain(ua); e = ble_rx_get(t[1], b); CHECK(e && e->len == 5); release(e); + unsigned before_writable = writable_count; hold = 1; for (int i = 0; i < BLE_QUEUE_MAX; i++) CHECK(!ble_tx_put(t[0], a, frame(1, i))); e = frame(1, 99); CHECK(ble_tx_put(t[0], a, e) == BLE_WOULD_BLOCK); CHECK(!e->owner && e->dgram[0] == 99); release(e); - hold = 0; schedule(backends[0]); drain(ua); CHECK(writable_count == 1); + hold = 0; schedule(backends[0]); drain(ua); CHECK(writable_count > before_writable); for (int i = 0; i < BLE_QUEUE_MAX; i++) { e = ble_rx_get(t[1], b); CHECK(e && e->dgram[0] == i); release(e); } CHECK(!ble_rx_get(t[1], b)); for (int i = 0; i < BLE_QUEUE_MAX + 1; i++) { CHECK(!ble_tx_put(t[0], a, frame(2, i))); drain(ua); }