RpcChannel.h

/*
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 *
 * RpcChannel.h
 * The RPC Channel
 * Copyright (C) 2005 Simon Newton
 */

#ifndef COMMON_RPC_RPCCHANNEL_H_
#define COMMON_RPC_RPCCHANNEL_H_

#if HAVE_CONFIG_H
#include <config.h>
#endif // HAVE_CONFIG_H

#include <stdint.h>
#include <google/protobuf/service.h>
#include <ola/Callback.h>
#include <ola/io/Descriptor.h>
#include <ola/util/SequenceNumber.h>
#include <memory>

#include "ola/ExportMap.h"

#include HASH_MAP_H

namespace ola {
namespace rpc {

class RpcMessage;
class RpcService;

class RpcChannel {
 public :
 typedef SingleUseCallback1<void, class RpcSession*> CloseCallback;

 RpcChannel(RpcService *service,
 ola::io::ConnectedDescriptor *descriptor,
 ExportMap *export_map = NULL);

 ~RpcChannel();

 void SetService(RpcService *service) { m_service = service; }

 bool PendingRPCs() const { return !m_requests.empty(); }

 void DescriptorReady();

 void SetChannelCloseHandler(CloseCallback *callback);

 void CallMethod(const google::protobuf::MethodDescriptor *method,
 class RpcController *controller,
 const google::protobuf::Message *request,
 google::protobuf::Message *response,
 SingleUseCallback0<void> *done);

 void RequestComplete(class OutstandingRequest *request);

 RpcSession *Session();

 static const unsigned int PROTOCOL_VERSION = 1;

 private:
 typedef HASH_NAMESPACE::HASH_MAP_CLASS<int, class OutstandingResponse*>
 ResponseMap;

 std::auto_ptr<RpcSession> m_session;
 RpcService *m_service; // service to dispatch requests to
 std::auto_ptr<CloseCallback> m_on_close;
 // the descriptor to read/write to.
 class ola::io::ConnectedDescriptor *m_descriptor;
 SequenceNumber<uint32_t> m_sequence;
 uint8_t *m_buffer; // buffer for incoming msgs
 unsigned int m_buffer_size; // size of the buffer
 unsigned int m_expected_size; // the total size of the current msg
 unsigned int m_current_size; // the amount of data read for the current msg
 HASH_NAMESPACE::HASH_MAP_CLASS<int, class OutstandingRequest*> m_requests;
 ResponseMap m_responses;
 ExportMap *m_export_map;
 UIntMap *m_recv_type_map;

 bool SendMsg(RpcMessage *msg);
 int AllocateMsgBuffer(unsigned int size);
 int ReadHeader(unsigned int *version, unsigned int *size) const;
 bool HandleNewMsg(uint8_t *buffer, unsigned int size);
 void HandleRequest(RpcMessage *msg);
 void HandleStreamRequest(RpcMessage *msg);

 // server end
 void SendRequestFailed(class OutstandingRequest *request);
 void SendNotImplemented(int msg_id);
 void DeleteOutstandingRequest(class OutstandingRequest *request);

 // client end
 void HandleResponse(RpcMessage *msg);
 void HandleFailedResponse(RpcMessage *msg);
 void HandleCanceledResponse(RpcMessage *msg);
 void HandleNotImplemented(RpcMessage *msg);

 void HandleChannelClose();

 static const char K_RPC_RECEIVED_TYPE_VAR[];
 static const char K_RPC_RECEIVED_VAR[];
 static const char K_RPC_SENT_ERROR_VAR[];
 static const char K_RPC_SENT_VAR[];
 static const char *K_RPC_VARIABLES[];
 static const char STREAMING_NO_RESPONSE[];
 static const unsigned int INITIAL_BUFFER_SIZE = 1 << 11; // 2k
 static const unsigned int MAX_BUFFER_SIZE = 1 << 20; // 1M
};
} // namespace rpc
} // namespace ola
#endif // COMMON_RPC_RPCCHANNEL_H_