Skip to content

Commit

Permalink
Add pause and stop to Websocket Server (#187)
Browse files Browse the repository at this point in the history 
* add stop and pause to websocket server

Signed-off-by: Mabel Zhang <mabel@openrobotics.org>

* better comments

Signed-off-by: Mabel Zhang <mabel@openrobotics.org>

* generalize sim pause/stop to a service operation

Signed-off-by: Mabel Zhang <mabel@openrobotics.org>

* Use RequestRaw

Signed-off-by: Nate Koenig <nate@openrobotics.org>

* Rename service to request

Signed-off-by: Nate Koenig <nate@openrobotics.org>

* Update to new request signature

Signed-off-by: Nate Koenig <natekoenig@gmail.com>

---------

Signed-off-by: Mabel Zhang <mabel@openrobotics.org>
Signed-off-by: Nate Koenig <nate@openrobotics.org>
Signed-off-by: Nate Koenig <natekoenig@gmail.com>
Co-authored-by: Nate Koenig <nate@openrobotics.org>
Co-authored-by: Nate Koenig <natekoenig@gmail.com>
  • Loading branch information
@mabelzhang @nkoenig
3 people authored Mar 9, 2023
1 parent cdc6270 commit be1ea4c
Show file tree
Hide file tree
Showing 2 changed files with 86 additions and 14 deletions.
63 changes: 59 additions & 4 deletions plugins/websocket_server/WebsocketServer.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -749,7 +749,7 @@ void WebsocketServer::OnMessage(int _socketId, const std::string _msg)
}
else if (frameParts[0] == "topics")
{
igndbg << "Topic list request recieved\n";
igndbg << "Topic list request received\n";
ignition::msgs::StringMsg_V msg;

std::vector<std::string> topics;
Expand All @@ -770,7 +770,7 @@ void WebsocketServer::OnMessage(int _socketId, const std::string _msg)
}
else if (frameParts[0] == "topics-types")
{
igndbg << "Topic and message type list request recieved\n";
igndbg << "Topic and message type list request received\n";
ignition::msgs::Publishers msg;

std::vector<std::string> topics;
Expand Down Expand Up @@ -800,7 +800,7 @@ void WebsocketServer::OnMessage(int _socketId, const std::string _msg)
}
else if (frameParts[0] == "worlds")
{
igndbg << "World info request recieved\n";
igndbg << "World info request received\n";
ignition::msgs::Empty req;
req.set_unused(true);

Expand All @@ -820,7 +820,7 @@ void WebsocketServer::OnMessage(int _socketId, const std::string _msg)
}
else if (frameParts[0] == "scene")
{
igndbg << "Scene info request recieved for world["
igndbg << "Scene info request received for world["
<< frameParts[1] << "]\n";
ignition::msgs::Empty req;
req.set_unused(true);
Expand Down Expand Up @@ -998,6 +998,61 @@ void WebsocketServer::OnMessage(int _socketId, const std::string _msg)
{
this->OnAsset(_socketId, frameParts);
}
else if (frameParts[0] == this->operations[REQUEST])
{
this->OnRequest(_socketId, frameParts);
}
}

//////////////////////////////////////////////////
void WebsocketServer::OnRequest(int _socketId,
const std::vector<std::string> &_frameParts)
{
std::string service = _frameParts[1];
std::string msgTypeName = _frameParts[2];
std::string msgData = _frameParts[3];

igndbg << "Calling service [" << service << "]\n";
bool result;
unsigned int timeout = 2000;

std::vector<transport::ServicePublisher> publishers;
this->node.ServiceInfo(service, publishers);

if (publishers.empty())
{
std::cerr << "Node::RequestRaw(): Error getting response type for "
<< "service [" << service << "]\n";

ignition::msgs::StringMsg msg;
msg.set_data("service_not_found");
std::string data = BUILD_MSG(this->operations[REQUEST], service,
msg.GetTypeName(), msg.SerializeAsString());

// Queue the message for delivery.
this->QueueMessage(this->connections[_socketId].get(),
data.c_str(), data.length());

return;
}

std::string repTypeName = publishers.front().RepTypeName();

std::string repStr;
bool executed = this->node.RequestRaw(service, msgData, msgTypeName,
repTypeName, timeout, repStr, result);
if (!executed)
{
ignerr << "Unable to call service [" << service << "]\n";
}

// Construct the response message
std::string data = BUILD_MSG(this->operations[REQUEST], service,
repTypeName, repStr);

// Queue the message for delivery.
this->QueueMessage(this->connections[_socketId].get(),
data.c_str(), data.length());
}

//////////////////////////////////////////////////
Expand Down
37 changes: 27 additions & 10 deletions plugins/websocket_server/WebsocketServer.hh
View file
Original file line number Diff line number Diff line change
Expand Up @@ -103,17 +103,27 @@ namespace ignition
/// 8. "asset": Get a file as a byte array from a running Gazebo
/// server. Set the payload to the file URI that is
/// being requested.
/// 9. "worlds": Get world info.
/// 10. "scene": Get scene info.
/// 11. "image": Subscribe to an image in the `topic_name` component.
/// 12. "throttle": Throttle a topic in the `topic_name` component by
/// the rate in the `payload` component.
/// 13. "req": Request a service, passing in the optional request
/// message. The payload should be a serialized
/// protobuf message. The response payload holds the
/// serialized protobuf response, if any.
///
/// The `topic_name` component is mandatory for the "sub", "pub", and
/// "unsub" operations. If present, it must be the name of an Ignition
/// The `topic_name` component is mandatory for the "sub", "pub", "unsub",
/// and "req" operations. If present, it must be the name of an Ignition
/// Transport topic.
///
/// The `message_type` component is mandatory for the "pub" operation. If
/// present it names the Ignition Message type, such as
/// The `message_type` component is mandatory for the "pub" and "req"
/// operations. If present it names the Ignition Message type, such as
/// "ignition.msgs.Clock".
///
/// The `payload` component is mandatory for the "pub" operation. If
/// present, it contains a serialized string of an Ignition Message.
/// The `payload` component is mandatory for the "pub" and "req"
/// operations. If present, it contains a serialized string of an
/// Ignition Message.
///
/// ## Example frames
///
Expand Down Expand Up @@ -186,8 +196,6 @@ namespace ignition
/// \param[in] _msg The incoming message.
public: void OnMessage(int _socketId, const std::string _msg);

public: void OnRequestMessage(int _socketId, const std::string &_msg);

/// \brief Check and update subscription count for a message type. If
/// a client has more subscriptions to a topic of a specified type than
/// the subscription limit, this will block subscription. On the other
Expand All @@ -207,6 +215,12 @@ namespace ignition
private: void OnAsset(int _socketId,
const std::vector<std::string> &_frameParts);

/// \brief Handles service requests.
/// \param[in] _socketId Id of the socket associated with the message.
/// \param[in] _frameParts The request message in frame parts.
private: void OnRequest(int _socketId,
const std::vector<std::string> &_frameParts);

private: ignition::transport::Node node;

private: bool run = true;
Expand Down Expand Up @@ -304,13 +318,16 @@ namespace ignition

/// \brief Get an asset as a byte array.
ASSET = 4,

/// \brief Request a service
REQUEST = 5,
};

/// \brief The set of valid operations, in string form. These values
/// \brief The set of valid operations, in string form. These values
/// can be sent in websocket message frames.
/// These valus must align with the `Operation` enum.
private: std::vector<std::string> operations{
"sub", "pub", "topics", "protos", "asset"};
"sub", "pub", "topics", "protos", "asset", "req"};

/// \brief Store publish headers for topics. This is here to improve
/// performance. Keys are topic names and values are frame headers.
Expand Down

0 comments on commit be1ea4c

Please sign in to comment.