StreamRpcChannel.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
 *
 * StreamRpcChannel.h
 * Interface for the Stream RPC Channel
 * Copyright (C) 2005-2008 Simon Newton
 */

#ifndef COMMON_RPC_STREAMRPCCHANNEL_H_
#define COMMON_RPC_STREAMRPCCHANNEL_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 HASH_MAP_H

namespace ola {
namespace rpc {

using google::protobuf::Message;
using google::protobuf::MethodDescriptor;
using google::protobuf::RpcChannel;
using google::protobuf::RpcController;
using google::protobuf::Service;

class RpcMessage;

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

    int id;
    RpcController *controller;
    Message *response;
};

class OutstandingResponse {
  /*
   * These are Requests on the client end that haven't completed yet.
   */
  public:
    OutstandingResponse() {}
    ~OutstandingResponse() {}

    int id;
    RpcController *controller;
    google::protobuf::Closure *callback;
    Message *reply;
};


class StreamRpcChannel: public RpcChannel {
  public :
    StreamRpcChannel(Service *service,
                     ola::io::ConnectedDescriptor *descriptor,
                     ExportMap *export_map = NULL);

    ~StreamRpcChannel();

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

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

    void DescriptorReady();

    void SetChannelCloseHandler(SingleUseCallback0<void> *callback);

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

    void RequestComplete(OutstandingRequest *request);

    static const unsigned int PROTOCOL_VERSION = 1;

  private:
    typedef HASH_NAMESPACE::HASH_MAP_CLASS<int, 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();

    Service *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 incomming 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_STREAMRPCCHANNEL_H_