Skip to content

Change to mqtt Pub_Sub_Samples #90

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

Closed
wants to merge 7 commits into from
Closed

Change to mqtt Pub_Sub_Samples #90

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
2df75de
Added a mutex to guarantee the bytebuf gets cleared before next publi…
KaibaLopez Jan 29, 2020
2af87f1
fixed basic_pub_sub and added the wait predicate
KaibaLopez Jan 29, 2020
5c6d056
added a lock guard for the publish, and reset the publishCompleted
KaibaLopez Jan 29, 2020
317ce57
Merge branch 'master' into pubSubSampleMutex
KaibaLopez Jan 29, 2020
47274d3
using only one mutex
KaibaLopez Jan 29, 2020
63fdf8f
Merge branch 'pubSubSampleMutex' of github.com:aws/aws-iot-device-sdk…
KaibaLopez Jan 29, 2020
2e9441b
changed formatting
KaibaLopez Jan 31, 2020
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
14 changes: 12 additions & 2 deletions samples/mqtt/basic_pub_sub/main.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,6 +237,8 @@ int main(int argc, char *argv[])
bool connectionSucceeded = false;
bool connectionClosed = false;
bool connectionCompleted = false;
bool publishCompleted = false;
bool subscribeAck = false;

/*
* This will execute when an mqtt connect has completed or failed.
Expand Down Expand Up @@ -328,14 +330,17 @@ int main(int argc, char *argv[])
{
fprintf(stdout, "Subscribe failed with error %s\n", aws_error_debug_str(errorCode));
}
subscribeAck = true;
std::lock_guard<std::mutex> lockGuard(mutex);
conditionVariable.notify_one();
};

connection->Subscribe(topic.c_str(), AWS_MQTT_QOS_AT_LEAST_ONCE, onPublish, onSubAck);
conditionVariable.wait(uniqueLock);
conditionVariable.wait(uniqueLock, [&]() { return subscribeAck; });

while (true)
{
publishCompleted = false;
String input;
fprintf(
stdout,
Expand All @@ -352,7 +357,7 @@ int main(int argc, char *argv[])
ByteBuf payload = ByteBufNewCopy(DefaultAllocator(), (const uint8_t *)input.data(), input.length());
ByteBuf *payloadPtr = &payload;

auto onPublishComplete = [payloadPtr](Mqtt::MqttConnection &, uint16_t packetId, int errorCode) {
auto onPublishComplete = [&, payloadPtr](Mqtt::MqttConnection &, uint16_t packetId, int errorCode) {
aws_byte_buf_clean_up(payloadPtr);

if (packetId)
Expand All @@ -363,8 +368,13 @@ int main(int argc, char *argv[])
{
fprintf(stdout, "Operation failed with error %s\n", aws_error_debug_str(errorCode));
}
std::lock_guard<std::mutex> lockGuard(mutex);
publishCompleted = true;
conditionVariable.notify_one();
};

connection->Publish(topic.c_str(), AWS_MQTT_QOS_AT_LEAST_ONCE, false, payload, onPublishComplete);
conditionVariable.wait(uniqueLock, [&]() { return publishCompleted; });
}

/*
Expand Down
13 changes: 11 additions & 2 deletions samples/mqtt/raw_pub_sub/main.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -291,6 +291,8 @@ int main(int argc, char *argv[])
bool connectionSucceeded = false;
bool connectionClosed = false;
bool connectionCompleted = false;
bool publishCompleted = false;
bool subscribeAck = false;

/*
* This will execute when an mqtt connect has completed or failed.
Expand Down Expand Up @@ -382,14 +384,17 @@ int main(int argc, char *argv[])
{
fprintf(stdout, "Subscribe failed with error %s\n", aws_error_debug_str(errorCode));
}
subscribeAck = true;
std::lock_guard<std::mutex> lockGuard(mutex);
conditionVariable.notify_one();
};

connection->Subscribe(topic.c_str(), AWS_MQTT_QOS_AT_LEAST_ONCE, onPublish, onSubAck);
conditionVariable.wait(uniqueLock);
conditionVariable.wait(uniqueLock, [&]() { return subscribeAck; });

while (true)
{
publishCompleted = false;
String input;
fprintf(
stdout,
Expand All @@ -406,7 +411,7 @@ int main(int argc, char *argv[])
ByteBuf payload = ByteBufNewCopy(DefaultAllocator(), (const uint8_t *)input.data(), input.length());
ByteBuf *payloadPtr = &payload;

auto onPublishComplete = [payloadPtr](Mqtt::MqttConnection &, uint16_t packetId, int errorCode) {
auto onPublishComplete = [&, payloadPtr](Mqtt::MqttConnection &, uint16_t packetId, int errorCode) {
aws_byte_buf_clean_up(payloadPtr);

if (packetId)
Expand All @@ -417,8 +422,12 @@ int main(int argc, char *argv[])
{
fprintf(stdout, "Operation failed with error %s\n", aws_error_debug_str(errorCode));
}
std::lock_guard<std::mutex> lockGuard(mutex);
publishCompleted = true;
conditionVariable.notify_one();
};
connection->Publish(topic.c_str(), AWS_MQTT_QOS_AT_LEAST_ONCE, false, payload, onPublishComplete);
conditionVariable.wait(uniqueLock, [&]() { return publishCompleted; });
}

/*
Expand Down