CONNECTION = 'C'
AUTH = 'A'
ERROR = 'X'
EVENT = 'E'
RECORD = 'R'
RPC = 'P'
PRESENCE = 'U'
PRIVATE = 'PRIVATE/' 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): 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' CLOSED = 'CLOSED'
AWAITING_AUTHENTICATION = 'AWAITING_AUTHENTICATION'
AWAITING_CONNECTION = 'AWAITING_CONNECTION'
AUTHENTICATING = 'AUTHENTICATING'
OPEN = 'OPEN'
ERROR = 'ERROR'
RECONNECTING = 'RECONNECTING'
CHALLENGING = 'CHALLENGING' def parse(raw_messages, client):
def _parse_message(message, client):
def convert_typed(value, client): def get_message(topic, action, data=None):
def typed(value): class ResubscribeNotifier(object):
    """
    Makes sure that all functionality is resubscribed on reconnect. Subscription
    is called when the connection drops - which seems counterintuitive, but in
    fact just means that the re-subscription message will be added to the queue
    of messages that need re-sending as soon as the connection is
    re-established.

    Resubscribe logic should only occur once per connection loss.
    """

    def __init__(self, client, resubscribe):
        """
        Args:
            client: The deepstream client
            resubscribe: callable to call to allow resubscribing
        """
        self._client = client
        self._resubscribe = resubscribe

        self._is_reconnecting = False
        self._client.on(event_constants.CONNECTION_STATE_CHANGED,
                        self._handle_connection_state_changes)

    def destroy(self):
        self._client.remove_listener(event_constants.CONNECTION_STATE_CHANGED,
                                     self._handle_connection_state_changes)
        self._client = None

    def _handle_connection_state_changes(self, state):
        if state == connection_state.RECONNECTING and not self._is_reconnecting:
            self._is_reconnecting = True
        elif state == connection_state.OPEN and self._is_reconnecting:
            self._is_reconnecting = False
            self._resubscribe() class SingleNotifier(object):
    def __init__(self, client, connection, topic, action, timeout_duration):
        self._client = client
        self._connection = connection
        self._topic = topic
        self._action = action
        self._timeout_duration = timeout_duration

        self._requests = dict()

        self._resubscribe_notifier = ResubscribeNotifier(
            client, self._resend_requests)

    def has_request(self, name):
        return name in self._requests

    def request(self, name, callback):
        if name not in self._requests:
            self._requests[name] = []
            future = self._connection.send_message(self._topic, self._action,
                                                   [name])
        else:
            future = concurrent.Future()
            future.set_result()

        response_timeout = self._client.io_loop.call_later(
            self._timeout_duration, partial(self._on_response_timeout, name))
        self._requests[name].append({
            'timeout': response_timeout,
            'callback': callback
        })

        return future

    def receive(self, name, error, data):
        entries = self._requests[name]
        for entry in entries:
            self._client.io_loop.remove_timeout(entry['timeout'])
            entry['callback'](error, data)
        del self._requests[name]

    def _on_response_timeout(self, name):
        msg = ('No response received in time for '
               '{0}|{1}|{2}').format(self._topic, self._action, name)
        self._client._on_error(self._topic, event_constants.RESPONSE_TIMEOUT,
                               msg)

    def _resend_requests(self):
        for request in self._requests:
            self._connection.send_message(self._topic, self._action,
                                          [self._requests[request]]) class Listener(object):
    def __init__(self, listener_type, pattern, callback, options, client,
                 connection):
        self._type = listener_type
        self._callback = callback
        self._pattern = pattern
        self._options = options
        self._client = client
        self._connection = connection
        self._send_future = None

        subscription_timeout = options.get("subscriptionTimeout", 15)
        self._ack_timeout = connection._io_loop.call_later(
            subscription_timeout, self._on_ack_timeout)
        self._resubscribe_notifier = ResubscribeNotifier(
            client, self._send_listen)
        self._send_listen()
        self._destroy_pending = False

    def send_destroy(self):
        self._destroy_pending = True
        future = self._connection.send_message(
            self._type, action_constants.UNLISTEN, [self._pattern])
        self._resubscribe_notifier.destroy()
        return future

    def destroy(self):
        self._callback = None
        self._pattern = None
        self._client = None
        self._connection = None

    def accept(self, name):
        return self._connection.send_message(
            self._type, action_constants.LISTEN_ACCEPT, [self._pattern, name])

    def reject(self, name):
        return self._connection.send_message(
            self._type, action_constants.LISTEN_REJECT, [self._pattern, name])

    def _create_callback_response(self, message):
        return CallbackResponse(
            accept=partial(self.accept, message['data'][1]),
            reject=partial(self.reject, message['data'][1]))

    def _on_message(self, message):
        action = message['action']
        data = message['data']
        if action == action_constants.ACK:
            self._connection._io_loop.remove_timeout(self._ack_timeout)
        elif action == action_constants.SUBSCRIPTION_FOR_PATTERN_FOUND:
            # TODO: Show deprecated message
            self._callback(data[1], True,
                           self._create_callback_response(message))
        elif action == action_constants.SUBSCRIPTION_FOR_PATTERN_REMOVED:
            self._callback(data[1], False)
        else:
            is_found = (message['action'] ==
                        action_constants.SUBSCRIPTION_FOR_PATTERN_FOUND)
            self._callback(message['data'][1], is_found)

    def _send_listen(self):
        self._send_future = self._connection.send_message(
            self._type, action_constants.LISTEN, [self._pattern])

    def _on_ack_timeout(self):
        self._client._on_error(
            self._type, event_constants.ACK_TIMEOUT,
            ('No ACK message received in time for ' + self._pattern))

    @property
    def destroy_pending(self):
        return self._destroy_pending

    @property
    def send_future(self):
        return self._send_future 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(): def remote_wins(record, remote_value, remote_version, callback):
def local_wins(record, remote_value, remote_version, callback): SPLIT_REG_EXP = r"[\.\[\]]"
def get(data, path, deep_copy):
def set(data, path, value, deep_copy):
def _tokenize(path):
def _pad_list(_list, index, value):
from deepstreampy.constants import topic as topic_constants
from deepstreampy.constants import actions as action_constants
from deepstreampy.constants import event as event_constants
from deepstreampy.constants import connection_state
from deepstreampy.message import message_parser, message_builder
from deepstreampy.utils import ResubscribeNotifier, SingleNotifier, Listener
from deepstreampy.utils import str_types
from deepstreampy.constants import merge_strategies
from deepstreampy import jsonpath
from pyee import EventEmitter
from tornado import gen, concurrent
from functools import partial
from copy import deepcopy
import json
from __future__ import absolute_import, division, print_function, with_statement
from __future__ import unicode_literals




ALL_EVENT = 'ALL_EVENT'
ENTRY_ADDED_EVENT = 'ENTRY_ADDED_EVENT'
ENTRY_REMOVED_EVENT = 'ENTRY_REMOVED_EVENT'
ENTRY_MOVED_EVENT = 'ENTRY_MOVED_EVENT'


class Record(EventEmitter, object):
    def __init__(self, name, connection, options, client):
        super(Record, self).__init__()
        self.name = name
        self.usages = 0
        self._connection = connection
        self._client = client
        self._options = options

        self._has_provider = False
        self._is_ready = False
        self._is_destroyed = False
        self._data = {}
        self._version = None
        self._old_value = None
        self._old_path_values = None
        self._queued_method_calls = list()
        self._write_callbacks = {}
        self.merge_strategy = merge_strategies.remote_wins

        self._emitter = EventEmitter()

        if 'merge_strategy' in options:
            self.merge_strategy = options['merge_strategy']

        self._resubscribe_notifier = ResubscribeNotifier(
            client, self._send_read)
        record_read_ack_timeout = options.get("recordReadAckTimeout", 15)
        self._read_ack_timeout = client.io_loop.call_later(
            record_read_ack_timeout,
            partial(self._on_timeout, event_constants.ACK_TIMEOUT))

        record_read_timeout = options.get("recordReadTimeout", 15)
        self._read_timeout = client.io_loop.call_later(
            record_read_timeout,
            partial(self._on_timeout, event_constants.RESPONSE_TIMEOUT))

        self._record_delete_timeout = options.get("recordDeleteTimeout", 15)

        self._delete_ack_timeout = None
        self._discard_timeout = None

    def get(self, path=None):
        """
        Returns a copy of either the entire dataset of the record or, if called
        with a path - the value of that path within the record's dataset.

        Returning a copy rather than the actual value helps to prevent the
        record getting out of sync due to unintentional changes to its data.

        Args:
            path (str, optional): a JSON path
        """
        return self._get_path(path)

    def set(self, data, path=None, callback=None):
        """
        Sets the value of either the entire dataset or of a specific path within
        the record and submits the changes to the server.

        If the new data is equal to the current data, nothing happens.

        Args:
            data: the new value of the data
            path (str, optional): a JSON path
            callback (callable)
        """
        config = {}
        if callback:
            state = self._client.connection_state
            if state in (connection_state.CLOSED,
                         connection_state.RECONNECTING):

                callback('Connection error: error updating record as '
                         'connection was closed')
                return
            else:
                config['writeSuccess'] = True
                self._set_up_callback(self.version, callback)

        if path is None and not isinstance(data, (dict, list)):
            raise ValueError(
                "Invalid record data {0}: Record data must be a dict or list.")

        if self._check_destroyed('set'):
            return

        if not self._is_ready:
            self._queued_method_calls.append(partial(self.set, data, path))
            return

        old_value = self._data
        deep_copy = self._options.get('recordDeepCopy', True)

        new_value = jsonpath.set(old_value, path, data, deep_copy)