# Path: deepstreampy/constants/topic.py
# CONNECTION = 'C'
# AUTH = 'A'
# ERROR = 'X'
# EVENT = 'E'
# RECORD = 'R'
# RPC = 'P'
# PRESENCE = 'U'
# PRIVATE = 'PRIVATE/'
# 
# Path: deepstreampy/constants/actions.py
# PING = 'PI'
# PONG = 'PO'
# ACK = 'A'
# REDIRECT = 'RED'
# CHALLENGE = 'CH'
# CHALLENGE_RESPONSE = 'CHR'
# READ = 'R'
# CREATE = 'C'
# UPDATE = 'U'
# PATCH = 'P'
# DELETE = 'D'
# SUBSCRIBE = 'S'
# UNSUBSCRIBE = 'US'
# HAS = 'H'
# SNAPSHOT = 'SN'
# INVOKE = 'I'
# SUBSCRIPTION_FOR_PATTERN_FOUND = 'SP'
# SUBSCRIPTION_FOR_PATTERN_REMOVED = 'SR'
# SUBSCRIPTION_HAS_PROVIDER = 'SH'
# LISTEN = 'L'
# UNLISTEN = 'UL'
# LISTEN_ACCEPT = 'LA'
# LISTEN_REJECT = 'LR'
# PROVIDER_UPDATE = 'PU'
# QUERY = 'Q'
# CREATEORREAD = 'CR'
# EVENT = 'EVT'
# ERROR = 'E'
# REQUEST = 'REQ'
# RESPONSE = 'RES'
# REJECTION = 'REJ'
# PRESENCE_JOIN = 'PNJ'
# PRESENCE_LEAVE = 'PNL'
# QUERY = 'Q'
# WRITE_ACKNOWLEDGEMENT = 'WA'
# def reverse_lookup(action):
# 
# Path: deepstreampy/constants/event.py
# CONNECTION_ERROR = 'connectionError'
# CONNECTION_STATE_CHANGED = 'connectionStateChanged'
# MAX_RECONNECTION_ATTEMPTS_REACHED = 'MAX_RECONNECTION_ATTEMPTS_REACHED'
# CONNECTION_AUTHENTICATION_TIMEOUT = 'CONNECTION_AUTHENTICATION_TIMEOUT'
# ACK_TIMEOUT = 'ACK_TIMEOUT'
# NO_RPC_PROVIDER = 'NO_RPC_PROVIDER'
# RESPONSE_TIMEOUT = 'RESPONSE_TIMEOUT'
# DELETE_TIMEOUT = 'DELETE_TIMEOUT'
# UNSOLICITED_MESSAGE = 'UNSOLICITED_MESSAGE'
# MESSAGE_DENIED = 'MESSAGE_DENIED'
# MESSAGE_PARSE_ERROR = 'MESSAGE_PARSE_ERROR'
# VERSION_EXISTS = 'VERSION_EXISTS'
# NOT_AUTHENTICATED = 'NOT_AUTHENTICATED'
# MESSAGE_PERMISSION_ERROR = 'MESSAGE_PERMISSION_ERROR'
# LISTENER_EXISTS = 'LISTENER_EXISTS'
# NOT_LISTENING = 'NOT_LISTENING'
# TOO_MANY_AUTH_ATTEMPTS = 'TOO_MANY_AUTH_ATTEMPTS'
# IS_CLOSED = 'IS_CLOSED'
# RECORD_NOT_FOUND = 'RECORD_NOT_FOUND'
# NOT_SUBSCRIBED = 'NOT_SUBSCRIBED'
# INVALID_RPC_CORRELATION_ID = 'INVALID_RPC_CORRELATION_ID'
# 
# Path: deepstreampy/message/message_builder.py
# def get_message(topic, action, data=None):
# def typed(value):
# 
# Path: deepstreampy/message/message_parser.py
# def parse(raw_messages, client):
# def _parse_message(message, client):
# def convert_typed(value, client):
# 
# Path: deepstreampy/utils.py
# class SingleNotifier(object):
# class Listener(object):
# class ResubscribeNotifier(object):
# class AckTimeoutRegistry(EventEmitter):
# class Undefined(object):
#     def __init__(self, client, connection, topic, action, timeout_duration):
#     def has_request(self, name):
#     def request(self, name, callback):
#     def receive(self, name, error, data):
#     def _on_response_timeout(self, name):
#     def _resend_requests(self):
#     def __init__(self, listener_type, pattern, callback, options, client,
#                  connection):
#     def send_destroy(self):
#     def destroy(self):
#     def accept(self, name):
#     def reject(self, name):
#     def _create_callback_response(self, message):
#     def _on_message(self, message):
#     def _send_listen(self):
#     def _on_ack_timeout(self):
#     def destroy_pending(self):
#     def send_future(self):
#     def __init__(self, client, resubscribe):
#     def destroy(self):
#     def _handle_connection_state_changes(self, state):
#     def __init__(self, client, topic, timeout_duration):
#     def add(self, name, action=None):
#     def remove(self, name, action=None):
#     def clear(self, message):
#     def _on_timeout(self, unique_name, name):
# def _pad_list(l, index, value):
#     def __repr__(self):
# def itoa(num, radix):
# def get_uid():


from deepstreampy.constants import topic as topic_constants
from deepstreampy.constants import actions
from deepstreampy.constants import event as event_constants
from deepstreampy.message import message_builder
from deepstreampy.message import message_parser
from deepstreampy import utils
from tornado import concurrent
from tornado import gen
from functools import partial
        if name in self._providers:
            raise ValueError("RPC {0} already registered".format(name))

        self._ack_timeout_registry.add(name, actions.SUBSCRIBE)
        self._providers[name] = callback

        return self._connection.send_message(topic_constants.RPC,
                                             actions.SUBSCRIBE,
                                             [name])

    def unprovide(self, name):
        if not name:
            raise ValueError("invalid argument name")

        if name in self._providers:
            del self._providers[name]
            self._ack_timeout_registry.add(name, actions.UNSUBSCRIBE)
            future = self._connection.send_message(topic_constants.RPC,
                                                   actions.UNSUBSCRIBE,
                                                   [name])
        else:
            future = concurrent.Future()
            future.set_result(None)

        return future

    @gen.coroutine
    def make(self, name, data):
        f = concurrent.Future()


# Path: deepstreampy/constants/topic.py
# CONNECTION = 'C'
# AUTH = 'A'
# ERROR = 'X'
# EVENT = 'E'
# RECORD = 'R'
# RPC = 'P'
# PRESENCE = 'U'
# PRIVATE = 'PRIVATE/'
# 
# Path: deepstreampy/constants/actions.py
# PING = 'PI'
# PONG = 'PO'
# ACK = 'A'
# REDIRECT = 'RED'
# CHALLENGE = 'CH'
# CHALLENGE_RESPONSE = 'CHR'
# READ = 'R'
# CREATE = 'C'
# UPDATE = 'U'
# PATCH = 'P'
# DELETE = 'D'
# SUBSCRIBE = 'S'
# UNSUBSCRIBE = 'US'
# HAS = 'H'
# SNAPSHOT = 'SN'
# INVOKE = 'I'
# SUBSCRIPTION_FOR_PATTERN_FOUND = 'SP'
# SUBSCRIPTION_FOR_PATTERN_REMOVED = 'SR'
# SUBSCRIPTION_HAS_PROVIDER = 'SH'
# LISTEN = 'L'
# UNLISTEN = 'UL'
# LISTEN_ACCEPT = 'LA'
# LISTEN_REJECT = 'LR'
# PROVIDER_UPDATE = 'PU'
# QUERY = 'Q'
# CREATEORREAD = 'CR'
# EVENT = 'EVT'
# ERROR = 'E'
# REQUEST = 'REQ'
# RESPONSE = 'RES'
# REJECTION = 'REJ'
# PRESENCE_JOIN = 'PNJ'
# PRESENCE_LEAVE = 'PNL'
# QUERY = 'Q'
# WRITE_ACKNOWLEDGEMENT = 'WA'
# def reverse_lookup(action):
# 
# Path: deepstreampy/constants/event.py
# CONNECTION_ERROR = 'connectionError'
# CONNECTION_STATE_CHANGED = 'connectionStateChanged'
# MAX_RECONNECTION_ATTEMPTS_REACHED = 'MAX_RECONNECTION_ATTEMPTS_REACHED'
# CONNECTION_AUTHENTICATION_TIMEOUT = 'CONNECTION_AUTHENTICATION_TIMEOUT'
# ACK_TIMEOUT = 'ACK_TIMEOUT'
# NO_RPC_PROVIDER = 'NO_RPC_PROVIDER'
# RESPONSE_TIMEOUT = 'RESPONSE_TIMEOUT'
# DELETE_TIMEOUT = 'DELETE_TIMEOUT'
# UNSOLICITED_MESSAGE = 'UNSOLICITED_MESSAGE'
# MESSAGE_DENIED = 'MESSAGE_DENIED'
# MESSAGE_PARSE_ERROR = 'MESSAGE_PARSE_ERROR'
# VERSION_EXISTS = 'VERSION_EXISTS'
# NOT_AUTHENTICATED = 'NOT_AUTHENTICATED'
# MESSAGE_PERMISSION_ERROR = 'MESSAGE_PERMISSION_ERROR'
# LISTENER_EXISTS = 'LISTENER_EXISTS'
# NOT_LISTENING = 'NOT_LISTENING'
# TOO_MANY_AUTH_ATTEMPTS = 'TOO_MANY_AUTH_ATTEMPTS'
# IS_CLOSED = 'IS_CLOSED'
# RECORD_NOT_FOUND = 'RECORD_NOT_FOUND'
# NOT_SUBSCRIBED = 'NOT_SUBSCRIBED'
# INVALID_RPC_CORRELATION_ID = 'INVALID_RPC_CORRELATION_ID'
# 
# Path: deepstreampy/message/message_builder.py
# def get_message(topic, action, data=None):
# def typed(value):
# 
# Path: deepstreampy/message/message_parser.py
# def parse(raw_messages, client):
# def _parse_message(message, client):
# def convert_typed(value, client):
# 
# Path: deepstreampy/utils.py
# class SingleNotifier(object):
# class Listener(object):
# class ResubscribeNotifier(object):
# class AckTimeoutRegistry(EventEmitter):
# class Undefined(object):
#     def __init__(self, client, connection, topic, action, timeout_duration):
#     def has_request(self, name):
#     def request(self, name, callback):
#     def receive(self, name, error, data):
#     def _on_response_timeout(self, name):
#     def _resend_requests(self):
#     def __init__(self, listener_type, pattern, callback, options, client,
#                  connection):
#     def send_destroy(self):
#     def destroy(self):
#     def accept(self, name):
#     def reject(self, name):
#     def _create_callback_response(self, message):
#     def _on_message(self, message):
#     def _send_listen(self):
#     def _on_ack_timeout(self):
#     def destroy_pending(self):
#     def send_future(self):
#     def __init__(self, client, resubscribe):
#     def destroy(self):
#     def _handle_connection_state_changes(self, state):
#     def __init__(self, client, topic, timeout_duration):
#     def add(self, name, action=None):
#     def remove(self, name, action=None):
#     def clear(self, message):
#     def _on_timeout(self, unique_name, name):
# def _pad_list(l, index, value):
#     def __repr__(self):
# def itoa(num, radix):
# def get_uid():


# Path: deepstreampy/rpc.py
from deepstreampy.constants import topic as topic_constants
from deepstreampy.constants import actions
from deepstreampy.constants import event as event_constants
from deepstreampy.message import message_builder
from deepstreampy.message import message_parser
from deepstreampy import utils
from tornado import concurrent
from tornado import gen
from functools import partial

        if name in self._providers:
            raise ValueError("RPC {0} already registered".format(name))

        self._ack_timeout_registry.add(name, actions.SUBSCRIBE)
        self._providers[name] = callback

        return self._connection.send_message(topic_constants.RPC,
                                             actions.SUBSCRIBE,
                                             [name])

    def unprovide(self, name):
        if not name:
            raise ValueError("invalid argument name")

        if name in self._providers:
            del self._providers[name]
            self._ack_timeout_registry.add(name, actions.UNSUBSCRIBE)
            future = self._connection.send_message(topic_constants.RPC,
                                                   actions.UNSUBSCRIBE,
                                                   [name])
        else:
            future = concurrent.Future()
            future.set_result(None)

        return future

    @gen.coroutine
    def make(self, name, data):
        f = concurrent.Future()


        uid = utils.get_uid()