metacall/mc_protocol.cpp

207 lines
5.9 KiB
C++
Raw Normal View History

2011-08-28 15:45:42 +00:00
//
// Copyright (c) 2011 Alex Yatskov
//
// Permission is hereby granted, free of charge, to any person
// obtaining a copy of this software and associated documentation
// files (the "Software"), to deal in the Software without
// restriction, including without limitation the rights to use,
// copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the
// Software is furnished to do so, subject to the following
// conditions:
//
// The above copyright notice and this permission notice shall be
// included in all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
// EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
// OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
// NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
// HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
// WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
// OTHER DEALINGS IN THE SOFTWARE.
//
#include "metacall.hpp"
namespace metacall {
//
// Protocol
//
Protocol::Protocol(Stream* stream, Binding* binding) :
2012-02-19 17:55:16 +00:00
m_stream(stream),
m_binding(binding),
m_rate(64)
2011-08-28 15:45:42 +00:00
{
}
void Protocol::advance() {
2012-02-20 02:47:09 +00:00
if (!m_stream->socket().connected()) {
2011-08-28 15:45:42 +00:00
return;
}
2012-02-19 17:55:16 +00:00
for (int i = 0; i < m_rate; ++i) {
2011-08-28 15:45:42 +00:00
if (advanceStream() != Stream::STATE_READY) {
break;
}
}
std::vector<TaskMap::iterator> processed;
2012-02-19 17:55:16 +00:00
for (TaskMap::iterator iter = m_taskMap.begin(); iter != m_taskMap.end(); ++iter) {
2011-08-28 15:45:42 +00:00
TaskEntry& entry = iter->second;
switch (entry.state) {
case TASK_STATE_READY:
case TASK_STATE_ERROR:
processed.push_back(iter);
for (size_t i = 0; i < entry.handlers.size(); ++i) {
HandlerEntry& handler = entry.handlers[i];
handler.handler(this, iter->first, entry.state, handler.userPtr);
}
break;
default:
break;
}
}
for (size_t i = 0; i < processed.size(); ++i) {
2012-02-19 17:55:16 +00:00
m_taskMap.erase(processed[i]);
2011-08-28 15:45:42 +00:00
}
}
void Protocol::setRate(int rate) {
2012-02-19 17:55:16 +00:00
m_rate = rate;
2011-08-28 15:45:42 +00:00
}
2011-09-17 17:09:28 +00:00
bool Protocol::pendingTasks() const {
2012-02-19 17:55:16 +00:00
return m_taskMap.size() > 0;
2011-09-17 17:09:28 +00:00
}
2011-08-28 15:45:42 +00:00
bool Protocol::setHandler(TaskId id, HandlerProc handler, void* userPtr) {
2012-02-19 17:55:16 +00:00
const TaskMap::iterator iter = m_taskMap.find(id);
if (iter == m_taskMap.end() || iter->second.state != TASK_STATE_PENDING) {
2011-08-28 15:45:42 +00:00
return false;
}
iter->second.handlers.push_back(HandlerEntry(handler, userPtr));
return true;
}
void Protocol::clearHandler(TaskId id) {
2012-02-19 17:55:16 +00:00
const TaskMap::iterator iter = m_taskMap.find(id);
if (iter != m_taskMap.end()) {
m_taskMap.erase(iter);
2011-08-28 15:45:42 +00:00
}
}
void Protocol::clearHandlers() {
2012-02-19 17:55:16 +00:00
for (TaskMap::iterator iter = m_taskMap.begin(); iter != m_taskMap.end(); ++iter) {
2011-08-28 15:45:42 +00:00
iter->second.handlers.clear();
}
}
Stream::State Protocol::advanceStream() {
2012-02-19 17:55:16 +00:00
m_stream->advance();
2011-08-28 15:45:42 +00:00
PacketHeader header;
int headerSize = 0;
2012-02-19 17:55:16 +00:00
Stream::State state = m_stream->peek(&header, &headerSize);
2011-08-28 15:45:42 +00:00
if (state == Stream::STATE_READY) {
switch (header.id) {
case PACKET_ID_FUNCTION_CALL_REQUEST:
state = advanceInvokeRequest();
break;
case PACKET_ID_FUNCTION_CALL_REPLY:
state = advanceInvokeReply();
break;
default:
state = Stream::STATE_ERROR_PROTOCOL;
break;
}
}
return state;
}
Stream::State Protocol::advanceInvokeReply() {
PacketInvokeReply packetReply;
Deserializer deserializer(&packetReply.data);
2012-02-19 17:55:16 +00:00
Stream::State state = m_stream->receive(&packetReply);
2011-08-28 15:45:42 +00:00
if (state == Stream::STATE_READY) {
2012-02-19 17:55:16 +00:00
const TaskMap::iterator iter = m_taskMap.find(static_cast<TaskId>(packetReply.taskId));
if (iter == m_taskMap.end()) {
2011-08-28 15:45:42 +00:00
state = Stream::STATE_ERROR_PROTOCOL;
}
else {
const bool error =
(packetReply.flags & PacketInvokeReply::FLAG_UNBOUND_FUNC) ||
(packetReply.flags & PacketInvokeReply::FLAG_INVALID_ARGS);
2011-08-28 15:45:42 +00:00
iter->second.state = error ? TASK_STATE_ERROR : TASK_STATE_READY;
iter->second.data = packetReply.data;
}
}
return state;
}
Stream::State Protocol::advanceInvokeRequest() {
PacketInvokeRequest packetRequest;
Deserializer deserializer(&packetRequest.data);
2012-02-19 17:55:16 +00:00
const Stream::State state = m_stream->receive(&packetRequest);
2011-08-28 15:45:42 +00:00
if (state == Stream::STATE_READY) {
PacketInvokeReply packetReply;
packetReply.taskId = packetRequest.taskId;
Serializer serializer(&packetReply.data);
2012-02-19 17:55:16 +00:00
switch (m_binding->call(packetRequest.function, &deserializer, &serializer)) {
2011-08-28 15:45:42 +00:00
case Binding::CALL_RESULT_INVALID_ARGS:
packetReply.flags |= PacketInvokeReply::FLAG_INVALID_ARGS;
break;
case Binding::CALL_RESULT_UNBOUND_FUNC:
packetReply.flags |= PacketInvokeReply::FLAG_UNBOUND_FUNC;
break;
default:
break;
}
2012-02-19 17:55:16 +00:00
m_stream->send(packetReply);
2011-08-28 15:45:42 +00:00
}
return state;
}
2011-09-05 03:05:33 +00:00
Protocol::TaskId Protocol::invoke(const Token& token) {
2011-08-28 15:45:42 +00:00
PacketInvokeRequest packetRequest;
return invokeExec(token, &packetRequest);
}
2011-09-05 03:05:33 +00:00
Protocol::TaskId Protocol::invokeExec(const Token& token, PacketInvokeRequest* packetRequest) {
2011-08-28 15:45:42 +00:00
const TaskId taskId = registerTaskId();
packetRequest->function = token;
packetRequest->taskId = taskId;
2012-02-19 17:55:16 +00:00
if (m_stream->send(*packetRequest) != Stream::STATE_READY) {
2011-08-28 15:45:42 +00:00
return TASK_ID_INVALID;
}
2012-02-19 17:55:16 +00:00
m_taskMap[taskId] = TaskEntry();
2011-08-28 15:45:42 +00:00
return taskId;
}
2011-09-05 03:05:33 +00:00
Protocol::TaskId Protocol::registerTaskId() {
2012-02-19 17:55:16 +00:00
static int s_id = TASK_ID_INVALID;
while (++s_id == TASK_ID_INVALID);
return static_cast<TaskId>(s_id);
2011-08-28 15:45:42 +00:00
}
}