Skip to content

Commit 427f34a

Browse files
markarenmarkaren
committed
mqtt ws wip
1 parent 40efa01 commit 427f34a

5 files changed

Lines changed: 168 additions & 35 deletions

File tree

‎src/simple_socket/mqtt/MQTTBroker.cpp‎

Lines changed: 90 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,13 +3,15 @@
33

44
#include "simple_socket/TCPSocket.hpp"
55
#include "simple_socket/mqtt/mqtt_common.hpp"
6+
#include "simple_socket/ws/WebSocket.hpp"
67

78
#include <iostream>
89
#include <mutex>
910
#include <thread>
1011
#include <unordered_map>
1112
#include <unordered_set>
1213
#include <vector>
14+
#include <queue>
1315

1416

1517
using namespace simple_socket;
@@ -18,16 +20,19 @@ using namespace simple_socket;
1820
struct MQTTBroker::Impl {
1921

2022
explicit Impl(int port)
21-
: server_(port), stop_(false) {}
23+
: server_(port), ws_(port + 1), stop_(false) {}
2224

2325
void start() {
2426
listener_ = std::thread([this] { acceptLoop(); });
27+
wsListener_ = std::thread([this] { wsAcceptLoop(); });
2528
}
2629

2730
void stop() {
2831
stop_ = true;
2932
server_.close();
33+
ws_.stop();
3034
if (listener_.joinable()) listener_.join();
35+
if (wsListener_.joinable()) wsListener_.join();
3136
}
3237

3338
private:
@@ -38,8 +43,10 @@ struct MQTTBroker::Impl {
3843
};
3944

4045
TCPServer server_;
46+
WebSocket ws_;
4147
std::atomic<bool> stop_;
4248
std::thread listener_;
49+
std::thread wsListener_;
4350

4451
std::mutex subsMutex_;
4552
std::unordered_map<std::string, std::vector<Client*>> subscribers_;
@@ -64,6 +71,88 @@ struct MQTTBroker::Impl {
6471
}
6572
}
6673

74+
void wsAcceptLoop() {
75+
76+
struct WsWrapper: SimpleConnection {
77+
78+
WebSocketConnection* connection;
79+
80+
explicit WsWrapper(WebSocketConnection* c): connection(c) {}
81+
82+
int read(uint8_t* buffer, size_t size) override {
83+
std::unique_lock lock(m_);
84+
cv_.wait(lock, [&]{ return closed_ || !queue_.empty(); });
85+
if (queue_.empty()) return 0; // closed and no data
86+
87+
std::string msg = std::move(queue_.front());
88+
queue_.pop_front();
89+
90+
size_t toCopy = std::min(size, msg.size());
91+
std::memcpy(buffer, msg.data(), toCopy);
92+
93+
if (toCopy < msg.size()) {
94+
// put the remainder back to the front so next read continues it
95+
queue_.push_front(msg.substr(toCopy));
96+
}
97+
98+
return static_cast<int>(toCopy);
99+
}
100+
101+
bool write(const uint8_t* data, size_t size) override {
102+
return connection->send(data, size);
103+
}
104+
105+
void close() override {
106+
{
107+
std::lock_guard lock(m_);
108+
closed_ = true;
109+
}
110+
cv_.notify_all();
111+
}
112+
113+
void push_back(const std::string& msg) {
114+
{
115+
std::lock_guard lock(m_);
116+
queue_.push_back(msg);
117+
}
118+
cv_.notify_one();
119+
}
120+
121+
private:
122+
bool closed_{false};
123+
std::deque<std::string> queue_;
124+
std::mutex m_;
125+
std::condition_variable cv_;
126+
};
127+
128+
std::unordered_map<WebSocketConnection*, WsWrapper*> connections;
129+
130+
ws_.onOpen = [this, &connections](WebSocketConnection* conn) {
131+
std::cout << "MQTTBroker: new WebSocket connection" << std::endl;
132+
auto client = std::make_unique<Client>();
133+
Client* clientPtr = client.get();
134+
clients_.push_back(clientPtr);
135+
auto wrapper = std::make_unique<WsWrapper>(conn);
136+
connections[conn] = wrapper.get();
137+
client->conn = std::move(wrapper);
138+
139+
std::thread(&Impl::handleClient, this, std::move(client)).detach();
140+
};
141+
ws_.onMessage = [&connections](WebSocketConnection* conn, const std::string& msg) {
142+
std::cout << "MQTTBroker: new message on WebSocket connection: " << msg << std::endl;
143+
connections[conn]->push_back(msg);
144+
};
145+
ws_.onClose = [](WebSocketConnection* conn) {
146+
std::cout << "MQTTBroker: WebSocket connection closed" << std::endl;
147+
};
148+
ws_.start();
149+
150+
while (!stop_) {
151+
std::this_thread::sleep_for(std::chrono::milliseconds(100));
152+
}
153+
ws_.stop();
154+
}
155+
67156
void handleClient(std::unique_ptr<Client> c) {
68157
try {
69158
// CONNECT

‎src/simple_socket/ws/WebSocket.cpp‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -62,8 +62,8 @@ struct WebSocket::Impl {
6262
auto ws = std::make_unique<WebSocketConnectionImpl>(callbacks, socket.accept(), WebSocketConnectionImpl::Role::Server);
6363
ws->run(handshake);
6464
connections.emplace_back(std::move(ws));
65-
} catch (std::exception&) {
66-
// std::cerr << ex.what() << std::endl;
65+
} catch (std::exception& ex) {
66+
std::cerr << ex.what() << std::endl;
6767
}
6868

6969
//cleanup connections

‎src/simple_socket/ws/WebSocketConnection.hpp‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -115,7 +115,6 @@ namespace simple_socket {
115115
std::mutex tx_mtx_;// serialize writes only
116116
std::atomic_bool closed_{false};
117117

118-
WebSocket* socket_{};
119118
std::unique_ptr<SimpleConnection> conn_;
120119
WebSocketCallbacks callbacks_;
121120
std::thread thread_;

‎tests/integration/mqtt_client.html‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
<!DOCTYPE html>
2+
<html lang="en">
3+
<head>
4+
<meta charset="UTF-8">
5+
<title>WS client</title>
6+
</head>
7+
<body>
8+
9+
<div class="slidecontainer">
10+
<input type="range" min="1" max="100" value="50" class="slider" id="myRange">
11+
<p id="output"></p>
12+
</div>
13+
14+
<script src="https://unpkg.com/mqtt/dist/mqtt.min.js"></script>
15+
16+
<script>
17+
const client = mqtt.connect("ws://127.0.0.1:1884");
18+
19+
client.on("connect", () => {
20+
console.log("✅ Connected to test.mosquitto.org");
21+
client.subscribe("test/topic", (err) => {
22+
if (!err) {
23+
console.log("📡 Subscribed to test/topic");
24+
// Publish a message
25+
client.publish("test/topic", "Hello from browser!");
26+
}
27+
});
28+
});
29+
30+
client.on("message", (topic, message) => {
31+
console.log(`📥 Message received on ${topic}: ${message.toString()}`);
32+
});
33+
34+
client.on("error", (err) => {
35+
console.log("❌ Connection error: " + err.message);
36+
});
37+
38+
client.on("close", () => {
39+
console.log("🔌 Connection closed");
40+
});
41+
42+
</script>
43+
44+
</body>
45+
</html>

‎tests/integration/run_mqtt_client_broker.cpp‎

Lines changed: 31 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -16,40 +16,40 @@ int main() {
1616
broker.start();
1717

1818
try {
19-
MQTTClient client("127.0.0.1", port, "SimpleSocketClient");
20-
client.connect(false);
21-
22-
std::string topic1 = "simple_socket/topic1";
23-
std::string topic2 = "simple_socket/topic2";
24-
25-
client.subscribe(topic1, [topic1](const auto& msg) {
26-
std::cout << "[" << topic1 << "] Got: " << msg << std::endl;
27-
});
28-
29-
client.subscribe(topic2, [topic2](const auto& msg) {
30-
std::cout << "[" << topic2 << "] Got: " << msg << std::endl;
31-
});
32-
33-
std::atomic_bool stop = false;
34-
std::thread([&client, &stop, topic1, topic2] {
35-
while (!stop) {
36-
std::this_thread::sleep_for(std::chrono::seconds(1));
37-
client.publish(topic1, "Hello from SimpleSocket MQTT!");
38-
client.publish(topic2, "Another hello from SimpleSocket MQTT!");
39-
}
40-
}).detach();
41-
42-
client.run();
43-
44-
std::thread([&client, topic2] {
45-
std::this_thread::sleep_for(std::chrono::seconds(3));
46-
client.unsubscribe(topic2);
47-
}).detach();
19+
// MQTTClient client("127.0.0.1", port, "SimpleSocketClient");
20+
// client.connect(false);
21+
//
22+
// std::string topic1 = "simple_socket/topic1";
23+
// std::string topic2 = "simple_socket/topic2";
24+
//
25+
// client.subscribe(topic1, [topic1](const auto& msg) {
26+
// std::cout << "[" << topic1 << "] Got: " << msg << std::endl;
27+
// });
28+
//
29+
// client.subscribe(topic2, [topic2](const auto& msg) {
30+
// std::cout << "[" << topic2 << "] Got: " << msg << std::endl;
31+
// });
32+
//
33+
// std::atomic_bool stop = false;
34+
// std::thread([&client, &stop, topic1, topic2] {
35+
// while (!stop) {
36+
// std::this_thread::sleep_for(std::chrono::seconds(1));
37+
// client.publish(topic1, "Hello from SimpleSocket MQTT!");
38+
// client.publish(topic2, "Another hello from SimpleSocket MQTT!");
39+
// }
40+
// }).detach();
41+
//
42+
// client.run();
43+
//
44+
// std::thread([&client, topic2] {
45+
// std::this_thread::sleep_for(std::chrono::seconds(3));
46+
// client.unsubscribe(topic2);
47+
// }).detach();
4848

4949
std::cout << "Press any key to exit..." << std::endl;
5050
std::cin.get();
51-
stop = true;
52-
client.close();
51+
// stop = true;
52+
// client.close();
5353
broker.stop();
5454

5555
} catch (const std::exception& e) {

0 commit comments

Comments
 (0)