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., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 * RpcChannel.h
 * The RPC Channel
 * Copyright (C) 2005-2008 Simon Newton
 */

#ifndef COMMON_RPC_RPCCHANNEL_H_
#define COMMON_RPC_RPCCHANNEL_H_

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

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

#include "ola/ExportMap.h"
#include "common/rpc/RpcController.h"

#include HASH_MAP_H

namespace ola {
namespace rpc {

class RpcMessage;
class RpcService;

class OutstandingRequest {
  /*
   * These are requests on the server end that haven't completed yet.
   */
 public:
    OutstandingRequest() {}
    ~OutstandingRequest() {}

    int id;
    RpcController *controller;
    google::protobuf::Message *response;
};

class RpcChannel {
 public :
    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(SingleUseCallback0<void> *callback);

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

    void RequestComplete(OutstandingRequest *request);

    static const unsigned int PROTOCOL_VERSION = 1;

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

    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(OutstandingRequest *request);
    void SendNotImplemented(int msg_id);
    void DeleteOutstandingRequest(OutstandingRequest *request);

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

    void HandleChannelClose();

    RpcService *m_service;  // service to dispatch requests to
    std::auto_ptr<SingleUseCallback0<void> > 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, OutstandingRequest*> m_requests;
    ResponseMap m_responses;
    ExportMap *m_export_map;
    UIntMap *m_recv_type_map;

    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_