text:
import os
import sys
import socket
import errno
import time
import struct
import logging
import asyncore

PY2 = sys.version_info[0] == 2
PY3 = sys.version_info[0] == 3
if PY3:
    BinaryType = bytearray
elif PY2:
    BinaryType = str
else:
    raise RuntimeError('Unsupported python version.')

try:
    import cPickle as pickle
except ImportError:
    import pickle

__version__ = '0.2.4'
__all__ = [
    "Error",
    "SessionS", "SessionC",
    "Server", "Client", "ClientInfinite",
    "run_once", "run_forever",
    "Sleep",
    "logger",
    "AsynMsgException", "MessageSizeOverflowError",
    "with_message_handler_config", "message_handler_config",
    "SessionKeepAliveParams",
    "MessagePacker", "MessagePacker_Pickle", "MessagePacker_Struct",
]

def _str_system_error(code):
    return '%d:%s' % (code, errno.errorcode.get(code, 'unknown'))

def _connect_with_timeout(sock, address, timeout):
    sock.setblocking(False)
    end_time = time.time() + timeout

    while True:
        time.sleep(0.001)
        code = sock.connect_ex(address)
        if code in (errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK) \
                or (code == errno.WSAEINVAL and os.name in ('nt', 'ce')):
            if end_time < time.time():
                return 1, None  # timeout
            else:
                continue
        if code in (0, errno.EISCONN):
            return 0, None  # connected
        else:
            return -1, code  # error

def _is_valid_message_format(msg):
    if not isinstance(msg, tuple):
        return False
    if len(msg) != 2:
        return False
    if msg[0] is None:
        return False
    return True

def _wrapper_asyncore_log(msg, type):
    if type == "info":
        logger.info(msg)
    elif type == "warning":
        logger.warn(msg)
    elif type == "error":
        logger.error(msg)
    else:
        raise NotImplementedError

logger = logging.getLogger("asynmsg")

class Sleep:
    def __init__(self, seconds):
        self.seconds = seconds

    def __call__(self, *args, **kwargs):
        time.sleep(self.seconds)

_runner_list = []

def _run_once(runner_list, extra_tick, use_poll):
    """
    :return True - success; False - extra_tick failure; runner object - the runner who tick failure
    """
    if len(runner_list) > 0:
        asyncore.loop(0, use_poll, None, 1)
        for runner in runner_list:
            if not runner.tick():
                return runner
    if extra_tick is not None:
        code = extra_tick()
        if code is False:
            return False
    return True

def run_once(runner_list=None, extra_tick=Sleep(0.001), use_poll=False, auto_stop=True):
    """
    :param auto_stop when tick error occur, stop all runners, except:
        if error was from a runner tick and the runner has set 'only_stop_self_when_tick_error' to True,
        then only this runner stop
    """
    if runner_list is None:
        runner_list = _runner_list

    code = _run_once(runner_list, extra_tick, use_poll)
    if code is True:  # no error
        return True
    elif code is False:  # extra tick error
        if auto_stop:
            for runner in list(runner_list):
                runner.stop()
        return False
    else:  # runner tick error
        if hasattr(code, 'only_stop_self_when_tick_error') and code.only_stop_self_when_tick_error:
            if auto_stop:
                code.stop()
            return True
        else:
            if auto_stop:
                for runner in list(runner_list):
                    runner.stop()
            return False
    return True

def run_forever(runner_list=None, extra_tick=Sleep(0.001), use_poll=False, auto_stop=True):
    if runner_list is None:
        runner_list = _runner_list

    while True:
        if not run_once(runner_list, extra_tick, use_poll, auto_stop):
            break

class Error:
    ERROR_OK = 0
    ERROR_SELECT = -1
    ERROR_REMOTE_CLOSED = -2
    ERROR_FORCE_CLOSE = -3
    ERROR_KEEP_ALIVE_TIMEOUT = -4
    ERROR_UNPACK_INVALID_MESSAGE_SIZE = -5
    ERROR_UNPACK_DECODE_MESSAGE = -6
    ERROR_RECV_MESSAGE_FORMAT = -7
    ERROR_HANDLE_MESSAGE = -8
    ERROR_CONNECT_SYSTEM = -9
    ERROR_CONNECT_TIMEOUT = -10
    ERROR_CONNECT_OPEN = -11
    ERROR_CONNECT_REFUSED = -12

    BASE_STR_ERROR_MAP = {
        ERROR_OK: 'ERROR_OK',
        ERROR_SELECT: 'ERROR_SELECT',
        ERROR_REMOTE_CLOSED: 'ERROR_REMOTE_CLOSED',
        ERROR_FORCE_CLOSE: 'ERROR_FORCE_CLOSE',
        ERROR_KEEP_ALIVE_TIMEOUT: 'ERROR_KEEP_ALIVE_TIMEOUT',
        ERROR_UNPACK_INVALID_MESSAGE_SIZE: 'ERROR_UNPACK_INVALID_MESSAGE_SIZE',
        ERROR_UNPACK_DECODE_MESSAGE: 'ERROR_UNPACK_DECODE_MESSAGE',
        ERROR_RECV_MESSAGE_FORMAT: 'ERROR_RECV_MESSAGE_FORMAT',
        ERROR_HANDLE_MESSAGE: 'ERROR_HANDLE_MESSAGE',
        ERROR_CONNECT_SYSTEM: 'ERROR_CONNECT_SYSTEM',
        ERROR_CONNECT_TIMEOUT: 'ERROR_CONNECT_TIMEOUT',
        ERROR_CONNECT_OPEN: 'ERROR_CONNECT_OPEN',
        ERROR_CONNECT_REFUSED: 'ERROR_CONNECT_REFUSED',
    }

    @staticmethod
    def str_error(code):
        return '%d:%s' % (code, Error.BASE_STR_ERROR_MAP.get(code, 'unknown'))

    def __init__(self):
        self._error = Error.ERROR_OK
        self._system_error = 0

    def __str__(self):
        s = Error.str_error(self.get_error())
        if self.has_system_error():
            s += ':'
            s += _str_system_error(self.get_system_error())
        return s

    def copy(self, other):
        self._error = other._error
        self._system_error = other._system_error

    def clear(self):
        self._error = Error.ERROR_OK
        self._system_error = 0

    def set_error(self, error, system_error=0):
        if self.has_error():
            return
        self._error = error
        self._system_error = system_error

    def get_error(self):
        return self._error

    def has_error(self):
        return self._error != Error.ERROR_OK

    def get_system_error(self):
        return self._system_error

    def has_system_error(self):
        return self._system_error != 0

class AsynMsgException(Exception):
    pass

class MessageSizeOverflowError(AsynMsgException):
    def __init__(self, msg_id, size, max_size):
        self.msg_id = msg_id
        self.size = size
        self.max_size = max_size

    def __str__(self):
        return 'MessageSizeOverflowError: msg_id=%s size=%d max_size=%d' % (self.msg_id, self.size, self.max_size)

class SessionKeepAliveParams:
    def __init__(self, idle_time=30, interval=10, probes=3, ping_id='keep_alive_ping', pong_id='keep_alive_pong'):
        self.idle_time = idle_time
        self.interval = interval
        self.probes = probes
        self.ping_id = ping_id
        self.pong_id = pong_id

class MessagePacker:
    def __init__(self, size_fmt='H'):
        self._size_fmt = size_fmt

    def pack(self, msg_id, msg_data):
        """ Pack to bytes, msg_data may be any type(including None)"""
        raise NotImplementedError

    def unpack(self, bytes):
        """Unpack to pair of (msg_id, msg_data)"""
        raise NotImplementedError

    @property
    def size_fmt(self):
        return self._size_fmt

class MessagePacker_Pickle(MessagePacker):
    def __init__(self, size_fmt='H'):
        super(MessagePacker_Pickle, self).__init__(size_fmt)

    def pack(self, msg_id, msg_data):
        return pickle.dumps((msg_id, msg_data))

    def unpack(self, bytes):
        return pickle.loads(bytes)

class MessagePacker_Struct(MessagePacker):
    def __init__(self, size_fmt='H', id_fmt='H'):
        super(MessagePacker_Struct, self).__init__(size_fmt)
        self._id_fmt = id_fmt

    def pack(self, msg_id, msg_data):
        bytes = struct.pack(self._id_fmt, msg_id)
        if msg_data is not None:
            bytes += msg_data
        return bytes

    def unpack(self, bytes):
        msg_id = struct.unpack_from(self._id_fmt, bytes)[0]
        msg_data = bytes[2:]
        return (msg_id, msg_data)

def with_message_handler_config(cls):
    # cls hasattr _command_factory if derived from a class which defined with_message_handler_config
    if not hasattr(cls, '_command_factory'):
        cls._command_factory = {}

        if cls.keep_alive_params is not None:
            def _on_keep_alive_ping(self, msg_id, msg_data):
                self.send_message(cls.keep_alive_params.pong_id, None)
            def _on_keep_alive_pong(self, msg_id, msg_data):
                pass
            cls.register_command_handler(cls.keep_alive_params.ping_id, _on_keep_alive_ping)
            cls.register_command_handler(cls.keep_alive_params.pong_id, _on_keep_alive_pong)

    order_map = {}

    for func in cls.__dict__.values():
        if hasattr(func,'_message_handler_index'):
            order_map[func._message_handler_index] = func

    # keys can't sort in python 3(the type is dict_keys)
    # so we first transform it to a list
    keys = order_map.keys()
    sort_keys = list(keys)
    sort_keys.sort()

    for k in sort_keys:
        func = order_map[k]
        cls.register_command_handler(func._message_handler_msg_id, func)

    return cls

class message_handler_config:
    total_count = 0

    def __init__(self, msg_id, allow_override=True):
        self.msg_id = msg_id
        self.allow_override = allow_override
        self.index = self.__class__.total_count
        self.__class__.total_count += 1

    def __call__(self, func):
        func._message_handler_index = self.index
        func._message_handler_msg_id = self.msg_id
        func._message_handler_allow_override = self.allow_override
        return func

class AsynMsgDispatcher(asyncore.dispatcher):
    def close(self):
        asyncore.dispatcher.close(self)
        self.socket = None

class _Session(AsynMsgDispatcher):
    message_packer = MessagePacker_Pickle()
    keep_alive_params = SessionKeepAliveParams()  # set None to disable
    max_message_size = 16 * 1024
    max_send_size_once = 16 * 1024
    max_recv_size_once = 16 * 1024
    enable_nagle_algorithm = False

    @classmethod
    def register_command_handler(cls, cmd, handler):
        if cmd in cls._command_factory:
            old_handler = cls._command_factory[cmd]
            if hasattr(old_handler,'_message_handler_allow_override') and old_handler._message_handler_allow_override:
                cls._command_factory[cmd] = handler
            else:
                raise RuntimeError("Can't register message handler with duplicate id '%s'." % cmd)
        cls._command_factory[cmd] = handler

    def __init__(self, sock, address):
        AsynMsgDispatcher.__init__(self, sock)

        self._manage_owner = None

        self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, not self.__class__.enable_nagle_algorithm)
        self._error = Error()

        self._in_buffer = BinaryType()
        self._out_buffer = BinaryType()

        self._last_read_time = time.time()
        self._keep_alive_probe_count = 0

        self._force_close_time = -1
        self._force_wait_timeout = False

        self._ready = True
        self._serial = -1

    def close(self):
        AsynMsgDispatcher.close(self)

    # close on no data to send or timeout, like linger
    # if force_wait_timeout is True, wait even if no data to send
    def force_close(self, timeout=5, force_wait_timeout=False):
        self._force_close_time = time.time() + timeout
        self._force_wait_timeout = force_wait_timeout

    def get_manage_owner(self):
        return self._manage_owner

    def set_ready(self, ready=True):
        self._ready = ready

    def is_ready(self):
        return self._ready

    def get_error(self):
        return self._error

    def get_local_address(self):
        return self.socket.getsockname()

    def get_remote_address(self):
        return self.addr

    def get_serial(self):
        return self._serial

    def get_low_level_desc(self):
        local_addr = self.get_local_address()
        remote_addr = self.get_remote_address()
        return '%s(%s:%d,%s:%d)(%d)' % (self.__class__.__name__, local_addr[0], local_addr[1], remote_addr[0], remote_addr[1], self._serial)

    def check_open(self):
        return True

    def on_opened(self):
        pass

    def on_closing(self):
        pass

    def on_closed(self):
        pass

    def tick(self):
        if not self._error.has_error():
            if self._force_close_time > 0:
                if (not self._force_wait_timeout and len(self._out_buffer) == 0) or self._force_close_time < time.time():
                    self._error.set_error(Error.ERROR_FORCE_CLOSE)

        if not self._error.has_error():
            self._keep_alive_check()

        if not self._error.has_error():
            self._unpack_and_handle_messages()

        return not self._error.has_error()

    """ <<< asyncore.dispatcher interfaces """
    def log(self, message):
        _wrapper_asyncore_log(message, 'info')

    def log_info(self, message, type='info'):
        _wrapper_asyncore_log(message, type)

    def readable(self):
        return not self._error.has_error()

    def writable(self):
        return (not self._error.has_error()) and len(self._out_buffer) > 0

    def handle_read(self):
        if self._error.has_error():
            return

        data = self.recv(self.__class__.max_recv_size_once)  # may handle_close inside
        if data:
            self._in_buffer += data
            self._last_read_time = time.time()
            self._keep_alive_probe_count = 0

    def handle_write(self):
        if self._error.has_error():
            return

        num = self.send(self._out_buffer[:self.__class__.max_send_size_once])
        if num > 0:
            self._out_buffer = self._out_buffer[num:]

    def handle_close(self):  # use set_error if want to close manually
        if self._error.has_error():
            return

        self._error.set_error(Error.ERROR_REMOTE_CLOSED)
    """ asyncore.dispatcher interfaces >>> """

    def handle_message(self, msg_id, msg_data):
        handler = self.__class__._command_factory.get(msg_id)
        if handler is None:
            return self.on_unhandled_message(msg_id, msg_data)
        else:
            return handler(self, msg_id, msg_data)

    def on_unhandled_message(self, msg_id, msg_data):
        logger.warning('%s unhandled message(%s)', self.get_low_level_desc(), msg_id)

    def send_message(self, msg_id, msg_data):
        """msg_data can be None"""
        if self._error.has_error() or self._force_close_time > 0:
            return False

        byte_msg = self.message_packer.pack(msg_id, msg_data)
        length = struct.calcsize(self.__class__.message_packer.size_fmt) + len(byte_msg)

        if length > self.__class__.max_message_size:
            raise MessageSizeOverflowError(msg_id, length, self.__class__.max_message_size)

        self._out_buffer += struct.pack(self.__class__.message_packer.size_fmt, length)
        self._out_buffer += byte_msg
        return True

    def _unpack_and_handle_messages(self):
        while True:
            pair = self._unpack_message()
            if pair in (0, -1):
                break
            if not _is_valid_message_format(pair):
                try:
                    logger.error('%s invalid message format: %s', self.get_low_level_desc(), str(pair))
                except:
                    pass
                self._error.set_error(Error.ERROR_RECV_MESSAGE_FORMAT)
                break

            code = self.handle_message(pair[0], pair[1])
            if code is False and not self._error.has_error():
                logger.error('%s handle message(%s) error', self.get_low_level_desc(), pair[0])
                self._error.set_error(Error.ERROR_HANDLE_MESSAGE)

            if self._error.has_error():
                break

    # 0: retry
    # -1: error
    def _unpack_message(self):
        size_length = struct.calcsize(self.__class__.message_packer.size_fmt)
        buff_length = len(self._in_buffer)
        if buff_length < size_length:
            return 0
        length = struct.unpack_from(self.__class__.message_packer.size_fmt, self._in_buffer)[0]
        if length < size_length or length > self.__class__.max_message_size:
            logger.error('%s invalid message size(%d), must between [%d~%d]', self.get_low_level_desc(), length, size_length, self.__class__.max_message_size)
            self._error.set_error(Error.ERROR_UNPACK_INVALID_MESSAGE_SIZE)
            return -1
        if buff_length < length:
            return 0
        byte_msg = self._in_buffer[size_length:length]

        try:
            pair = self.message_packer.unpack(byte_msg)
        except:
            self._error.set_error(Error.ERROR_UNPACK_DECODE_MESSAGE)
            return -1

        self._in_buffer = self._in_buffer[length:]
        return pair

    def _keep_alive_check(self):
        if self.__class__.keep_alive_params is None:
            return

        if self._keep_alive_probe_count > self.__class__.keep_alive_params.probes:
            return

        check_time = self.__class__.keep_alive_params.idle_time + \
                     self._keep_alive_probe_count * self.__class__.keep_alive_params.interval
        if time.time() - self._last_read_time > check_time:
            self._keep_alive_probe_count += 1
            if self._keep_alive_probe_count > self.__class__.keep_alive_params.probes:
                self._error.set_error(Error.ERROR_KEEP_ALIVE_TIMEOUT)
            else:
                self.send_message(self.__class__.keep_alive_params.ping_id, None)

    def _on_keep_alive_ping(self, msg_id, msg_data):
        self.send_message(self.__class__.keep_alive_params.pong_id, None)

class SessionS(_Session):
    def __init__(self, sock, address):
        _Session.__init__(self, sock, address)

    def on_opened(self):
        logger.info('%s open' % (self.get_low_level_desc()))

    def on_closing(self):
        logger.info('%s close, error(%s)' % (self.get_low_level_desc(), self._error))

class Server(AsynMsgDispatcher):
    session_class = SessionS
    only_stop_self_when_tick_error = False

    def __init__(self):
        AsynMsgDispatcher.__init__(self)
        self._session_map = {}
        self._next_serial = 0
        self._error = Error()

        self._listen_address = ''

    def set_listen_address(self, address):
        self._listen_address = address

    def start(self):
        assert not self.is_started()

        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.set_reuse_addr()
        self.bind(self._listen_address)
        self.listen(5)

        _runner_list.append(self)

        self.log_info('%s(%s:%d) start listening...' % (self.__class__.__name__, self._listen_address[0], self._listen_address[1]))

    def stop(self):
        if not self.is_started():
            return

        self.log_info('%s(%s:%d) stop listening' % (self.__class__.__name__, self._listen_address[0], self._listen_address[1]))

        _runner_list.remove(self)

        self._clear_sessions()
        self._error.clear()
        self.close()

    def tick(self):
        assert self.is_started()

        if self._error.has_error():  # by dispatcher
            return False

        for session in list(self._session_map.values()):  # make a list copy in case: session remove from map by _close_session
            if session.get_error().has_error():
                self._close_session(session)

        for session in self._session_map.values():
            session.tick()

        return True

    def is_started(self):
        return self.socket is not None

    """ <<< asyncore.dispatcher interfaces """
    def log(self, message):
        _wrapper_asyncore_log(message, 'info')

    def log_info(self, message, type='info'):
        _wrapper_asyncore_log(message, type)

    def handle_accepted(self, sock, address):
        self._open_session(sock, address)

    def handle_close(self):
        self._error.set_error(Error.ERROR_SELECT)
    """ asyncore.dispatcher interfaces >>> """

    def get_error(self):
        return self._error

    def find_session(self, serial):
        return self._session_map.get(serial)

    def get_sessions(self):
        result = []
        for session in self._session_map.values():
            result.append(session)
        return result

    def get_ready_sessions(self):
        result = []
        for session in self._session_map.values():
            if session.is_ready():
                result.append(session)
        return result

    def broadcast(self, msg_id, msg_data=None):
        for session in self._session_map.values():
            if session.is_ready():
                session.send_message(msg_id, msg_data)

    def check_session_open(self, session):
        return session.check_open()

    def on_session_opened(self, session):
        session.on_opened()

    def on_session_closing(self, session):
        session.on_closing()

    def on_session_closed(self, session):
        session.on_closed()

    def _clear_sessions(self):
        for session in list(self._session_map.values()):
            self._close_session(session)
        assert len(self._session_map) == 0  #self._session_map.clear()

    def _open_session(self, sock, address):
        session = self.__class__.session_class(sock, address)

        if not self.check_session_open(session):
            session.close()
            return False

        #{ build link
        session._manage_owner = self
        session._serial = self._next_serial
        self._next_serial += 1
        self._session_map[session.get_serial()] = session
        #}

        self.on_session_opened(session)
        return True

    def _close_session(self, session):
        self.on_session_closing(session)

        #{ break link
        del self._session_map[session.get_serial()]
        session._manage_owner = None
        #}

        session.close()
        self.on_session_closed(session)

class SessionC(_Session):
    def __init__(self, sock, address):
        _Session.__init__(self, sock, address)

    def on_opened(self):
        logger.info('%s connected' % (self.get_low_level_desc()))

    def on_closing(self):
        logger.info('%s disconnect, error(%s)' % (self.get_low_level_desc(), self._error))

class Client:
    session_class = SessionC
    only_stop_self_when_tick_error = False

    def __init__(self):
        self._started = False
        self._session = None
        self._next_serial = 0
        self._error = Error()

        self._connect_address = None
        self._connect_timeout = 5

    def set_connect_address(self, address):
        self._connect_address = address

    def set_connect_timeout(self, timeout):
        self._connect_timeout = timeout

    def _do_connect(self):
        sock = socket.socket()
        code, err = _connect_with_timeout(sock, self._connect_address, self._connect_timeout)

        if code == -1: # error
            self._error.set_error(Error.ERROR_CONNECT_SYSTEM, err)
            sock.close()
            return False

        if code == 1: # timeout
            self._error.set_error(Error.ERROR_CONNECT_TIMEOUT)
            sock.close()
            return False

        if not self._open_session(sock, self._connect_address):
            self._error.set_error(Error.ERROR_CONNECT_OPEN)
            return False

        return True

    def start(self):
        assert not self.is_started()

        self.log_info('%s(%s:%d) start connecting...' % (self.__class__.__name__, self._connect_address[0], self._connect_address[1]))
        if not self._do_connect():
            self.log_info('%s(%s:%d) connect failure(%s)' % (self.__class__.__name__, self._connect_address[0], self._connect_address[1], self._error))
            return False

        _runner_list.append(self)

        self._started = True
        return True

    def stop(self):
        if not self.is_started():
            return

        _runner_list.remove(self)

        self._close_session()
        self._error.clear()
        self._started = False

    def tick(self):
        assert self.is_started()

        if self._error.has_error():
            return False

        if self._session.get_error().has_error():
            self._error.copy(self._session.get_error())
            return False

        self._session.tick()
        return True

    def is_started(self):
        return self._started

    def get_error(self):
        return self._error

    def get_session(self):
        return self._session

    def get_ready_session(self):
        return self._session if (self._session is not None and self._session.is_ready()) else None

    def check_session_open(self, session):
        return session.check_open()

    def on_session_opened(self, session):
        session.on_opened()

    def on_session_closing(self, session):
        session.on_closing()

    def on_session_closed(self, session):
        session.on_closed()

    def _open_session(self, sock, address):
        session = self.__class__.session_class(sock, address)

        if not self.check_session_open(session):
            session.close()
            return False

        #{ build link
        session._manage_owner = self
        session._serial = self._next_serial
        self._next_serial += 1
        self._session = session
        #}

        self.on_session_opened(session)
        return True

    def _close_session(self):
        session = self._session

        self.on_session_closing(session)

        #{ break link
        self._session = None
        session._manage_owner = None
        #}

        session.close()
        self.on_session_closed(session)

    def log(self, message):
        _wrapper_asyncore_log(message, 'info')

    def log_info(self, message, type='info'):
        _wrapper_asyncore_log(message, type)

class ClientInfinite(AsynMsgDispatcher):
    session_class = SessionC
    only_stop_self_when_tick_error = False

    def __init__(self):
        AsynMsgDispatcher.__init__(self)
        self._started = False
        self._session = None
        self._next_serial = 0

        self._connect_address = None
        self._wait_retry_interval = 10

        self._connect_time = 0

    def set_connect_address(self, address):
        """在start前若没有调用该方法,则在start后调用该方法时会立即发起连接"""
        self._connect_address = address

    def set_wait_retry_interval(self, interval):
        self._wait_retry_interval = interval

    def start(self):
        assert not self.is_started()

        self.do_wait_retry(0)
        _runner_list.append(self)

        self._started = True
        return True

    def stop(self):
        if not self.is_started():
            return

        _runner_list.remove(self)

        if self._session is None:
            self.close()
        else:
            self._close_session()

        self._started = False

    def tick(self):
        assert self.is_started()

        if self._session is None:
            if self.socket is None:
                if self._connect_time < time.time():
                    if self._connect_address is not None:
                        self.log_info('%s(%s:%d) start connecting...' % (self.__class__.__name__, self._connect_address[0], self._connect_address[1]))
                        self.do_connect()
                else:
                    pass # 等待发起下次连接
            else:
                pass # 正在连接
        else:
            pass # 已经连接

        if self._session is not None:
            if self._session.get_error().has_error():
                self._close_session()

        if self._session is not None:
            self._session.tick()

        return True

    def wait_retry(self, interval=None):
        if interval is None:
            interval = self._wait_retry_interval
        if self._connect_address is not None:
            self.log_info('%s(%s:%d) try reconnect after %d seconds' % (self.__class__.__name__, self._connect_address[0], self._connect_address[1], interval))
        self.do_wait_retry(interval)

    def do_wait_retry(self, interval):
        """在interval秒后发起连接"""
        self._connect_time = time.time() + interval

    def do_connect(self):
        """发起连接"""
        assert self.socket is None
        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
        self.connect(self._connect_address)  # may handle_connect inside

    def is_started(self):
        return self._started

    def get_session(self):
        return self._session

    def get_ready_session(self):
        return self._session if (self._session is not None and self._session.is_ready()) else None

    def check_session_open(self, session):
        return session.check_open()

    def on_session_opened(self, session):
        session.on_opened()

    def on_session_closing(self, session):
        session.on_closing()

    def on_session_closed(self, session):
        session.on_closed()
        self.wait_retry()

    def _open_session(self, sock, address):
        session = self.__class__.session_class(sock, address)

        if not self.check_session_open(session):
            session.close()
            return False

        #{ build link
        session._manage_owner = self
        session._serial = self._next_serial
        self._next_serial += 1
        self._session = session
        #}

        self.on_session_opened(session)
        return True

    def _close_session(self):
        session = self._session

        self.on_session_closing(session)
        self._session = None
        session._manage_owner = None
        session.close()
        self.on_session_closed(session)

    """ <<< asyncore.dispatcher interfaces """
    def log(self, message):
        _wrapper_asyncore_log(message, 'info')

    def log_info(self, message, type='info'):
        _wrapper_asyncore_log(message, type)

    def handle_connect(self):
        #{ detach socket and forward to session
        self.del_channel()
        sock = self.socket
        self.socket = None
        #}

        if not self._open_session(sock, self.addr):
            self.wait_retry()
            return

    def handle_read(self):
        return  # the concrete session will handle_read

    def handle_write(self):
        return  # the concrete session will handle_write

    def handle_close(self):
        self.log_info('%s(%s:%d) connect failure, error(%s)' % (self.__class__.__name__, self._connect_address[0], self._connect_address[1],
                                                                _str_system_error(self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR))))
        self.close()
        self.wait_retry()
    """ asyncore.dispatcher interfaces >>> """


text:
title: Redis
category: page
slug: redis
sortorder: 0311
toc: False
sidebartitle: Redis
meta: Redis is an in-memory key-value pair data store classified as a NoSQL database and used with Python applications.

Redis is an in-memory key-value pair
database typically classified as a [NoSQL database](/no-sql-datastore.html).
Redis is commonly used for [caching](/caching.html), transient data storage
and as a holding area for data during analysis in Python applications.

Redis is an implementation of the NoSQL database concept. Learn more in the data chapter or view the table of contents for all topics.

### Redis tutorials
Redis is easy to install and start using compared to most other persistent
backends, but it's useful to follow a walkthrough if you have never
previously used Redis or any NoSQL data store.

* [How to Use Redis with Python 3 and redis-py on Ubuntu 16.04](/blog/install-redis-use-python-3-ubuntu-1604.html)
  contains detailed steps to install and start using Redis in Python.

* [How To Install and Use Redis](https://www.digitalocean.com/community/tutorials/how-to-install-and-use-redis)
  is a straightforward starter guide that includes installation instructions.

### Redis with Python
Redis is easier to use with Python if you have a code library client that
bridges from your code to your Redis instace. The following libraries and
resources provide more information on handling data in a Redis instance
with your Python code.

* [Redis-py](https://github.com/andymccurdy/redis-py) is a solid
  Python client to use with Redis.

* [Walrus](http://charlesleifer.com/blog/walrus-lightweight-python-utilities-for-working-with-redis/)
  is a higher-level Python wrapper for Redis with some caching, querying
  and data structure components build into the library.

* [Writing Redis in Python with Asyncio](http://jamesls.com/writing-redis-in-python-with-asyncio-part-1.html)
  shows a detailed example for how to use the new Asyncio standard library in
  Python 3.4+ for working with Redis. There is also a
  [EuroPython video of the talk](https://www.youtube.com/watch?v=CF8zt8l6SeI)
  that goes along with the code.

* [Cache_deco](https://github.com/alexk307/cache_deco) is a generic Python
  caching decorator library.

* [Write your own miniature Redis with Python](http://charlesleifer.com/blog/building-a-simple-redis-server-with-python/)
  doesn't actually use Redis but shows how you can write a simplified
  version of Redis' in-memory data store with Python. It's a good article
  to understand more about how NoSQL data stores can work under the covers.

* [Introduction to Redis streams with Python](http://charlesleifer.com/blog/redis-streams-with-python/)
  shows how to use the new (as of Redis version 5.0) append-only streams
  functionality via Python code.

### Redis tools and examples
Redis' wide applicability can be a downside if you don't know what to start
using it for in your application. The following code and posts provide common
use cases for Redis.

* [redis-labs-use-cases](https://github.com/Altoros/redis-labs-use-cases)
  has a couple of examples of using Redis to analyze geospatial data and
  tweets.

* [redis-migrate-tool](https://github.com/vipshop/redis-migrate-tool)
  is a library to make it easier to move data between redis clusters
  and groups.

* [redis-rdb-tools](https://github.com/sripathikrishnan/redis-rdb-tools)
  parses the Redis' database storage files and can dump the contents to
  JSON files.

### Redis Security
Redis should be customized out of its default configuration to secure it
against unauthorized and unauthenticated users. These resources provide
some advice on Reids security and guarding against data breaches.

* Redis, just as with any relational or NoSQL database, needs to be secured
  based on [security guidelines](http://www.antirez.com/news/96). There is
  also a post where the main author of Redis
  [cracks its security](http://www.antirez.com/news/96) to show the tradeoffs
  purposely made between ease of use and security in the default settings.

* [For God's sake, secure your Mongo/Redis/etc!](https://medium.com/@shahinism/for-gods-sake-secure-your-mongo-redis-etc-4f310cf1bed2)
  digs into the unfortunate default security settings that come with many
  NoSQL databases which can be used to compromise your systems. Make sure
  to not only install your dependencies such as Redis, but automate modifying
  default settings to lock them down against attackers.

### Specific Redis topics
Once you have configured Redis, become comfortable using it and locked it
down against malicious actors, you will want to learn more about operating,
scaling and collecting metrics. The following resources should help you
get started in those areas.

* [A Key Expired In Redis, You Won't Believe What Happened Next](https://engineering.grab.com/a-key-expired-in-redis-you-wont-believe-what-happened-next)
  is a well-written post on issues one team found with their configuration
  when they were using Redis for a lot of data caching.

* [Redis-playbook](https://github.com/mikeblum/redis-playbook) is an Ansible
  playbook for installing, configuring and securing a Redis instance.

* [Monitoring Redis](http://www.mikeperham.com/2017/04/20/monitoring-redis/)
  shows common commands for accessing meta data about your Redis databases,
  such as `info` and `slowlog`.

* GitHub wrote a retrospective on
  [moving persistent data out of Redis](http://githubengineering.com/moving-persistent-data-out-of-redis/)
  and into [MySQL](/mysql.html) that is worth a read as you scale up your
  Redis usage.

* [Learn Redis the hard way (in production)](https://tech.trivago.com/2017/01/25/learn-redis-the-hard-way-in-production/)
  investigates problems found with a development team's Redis
  infrastructure, how they went about debugging them and ultimately fixing
  some of the issues, while being aware of limitations that could cause
  them issues in the future.

* This video on
  [Scaling Redis at Twitter](https://www.youtube.com/watch?v=rP9EKvWt0zo) is
  a detailed look behind the scenes with a massive Redis deployment.

* [Real World Redis Tips](https://blog.heroku.com/real-world-redis-tips)
  provides some guidance from Heroku's engineers from deploying Redis at
  scale. The tips include setting an explicit idle connection timeout,
  using a connection pooler and avoiding using `KEYS` in favor of `SCAN`.

* [How to collect Redis metrics](https://www.datadoghq.com/blog/how-to-collect-redis-metrics/)
  shows how to use the Redis CLI client to grab key metrics on latency.

* [You should revise your Redis max connections setting](https://medium.com/appaloosa-store-engineering/you-should-revise-your-redis-max-connections-setting-8136f063c916)
  is a retrospective from a hard web application failure due to Redis
  connections maxing out on Heroku, and how to avoid this in your own
  applications by modifying your `redis.conf` settings.

* [A Speed Guide To Redis Lua Scripting](https://www.compose.com/articles/a-quick-guide-to-redis-lua-scripting/)
  shows how to use the Lua programming language to create extensions
  for Redis.

* [Better tests for Redis integrations with redislite](https://www.obeythetestinggoat.com/better-tests-for-redis-integrations-with-redislite.html)
  shows how to mock out a Redis instance using the
  [redislite](https://github.com/yahoo/redislite) library and clean up
  existing hacks you may be using to test your Redis usage.

* [Our journey from Redis 2 to Redis 3 while not taking the site down](https://engineering.skybettingandgaming.com/2017/09/25/redis-2-to-redis-3/)
  explains their infrastructure and uptime demands in the gambling industry,
  and how they were able to roll their upgraded Redis versions.

redpajama_set_name: RedPajamaGithub


text:
package com.liji.ssmtest.model;

/**
 * Created by Jimmy on 2016/9/23.
 */

public class User {
    private int user_id;
    private String user_name;
    private String user_age;

    public User(int id, String name, String age){
        super();
        this.setUser_id(id);
        this.setUser_name(name);
        this.setUser_age(age);

    }

    public User(){
        super();
    }

    public int getUser_id() {
        return user_id;
    }

    public void setUser_id(int user_id) {
        this.user_id = user_id;
    }

    public String getUser_name() {
        return user_name;
    }

    public void setUser_name(String user_name) {
        this.user_name = user_name;
    }

    public String getUser_age() {
        return user_age;
    }

    public void setUser_age(String user_age) {
        this.user_age = user_age;
    }
    @Override
    public String toString() {
        return "User{" +
                "user_name='" + user_name + '\'' +
                ", user_age='" + user_age + '\'' +
                ", user_id=" + user_id +
                '}';
    }
}


text:
import { applyMiddleware, createStore } from 'redux';
import thunk from 'redux-thunk';

import rootReducer from '../reducers';

export default function configureStore(initialState) {
  return createStore(
    rootReducer,
    initialState,
    applyMiddleware(thunk),
  );
}


text:
using NServiceBus;

#region AzureSharedHosting_HostConfiguration

public class EndpointConfig : IConfigureThisEndpoint, AsA_Host
{
    public void Customize(BusConfiguration configuration)
    {
    }
}

#endregion


text:
namespace Compute.Tests
{
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using Microsoft.Azure.Management.Compute;
    using Microsoft.Azure.Management.Compute.Models;
    using Microsoft.Azure.Management.ResourceManager;
    using Microsoft.Rest.ClientRuntime.Azure.TestFramework;
    using Xunit;

    public class VMScaleSetPriorityTests : VMScaleSetTestsBase
    {
        #region Priority Tests

        /// <summary>
        /// Covers following Operations:
        /// Create RG
        /// Create Storage Account
        /// Create Network Resources
        /// Get VMScaleSet Model View
        /// Get VMScaleSet Instance View
        /// List VMScaleSets in a RG
        /// List Available Skus
        /// Delete VMScaleSet
        /// Delete RG
        /// </summary>
        [Fact]
        [Trait("Name", "TestVMScaleSetScenarioOperations_Accept_Regular")]
        public void TestVMScaleSetPriorityOperations_Accept_Regular()
        {
            using (MockContext context = MockContext.Start(this.GetType()))
            {
                TestVMScaleSetPriorityOperationsInternal(context, VirtualMachinePriorityTypes.Regular);
            }
        }

        /// <summary>
        /// Covers following Operations:
        /// Create RG
        /// Create Storage Account
        /// Create Network Resources
        /// Create Low Priority VMScaleSet with no EvictionPolicy specified
        /// Get VMScaleSet Model View
        /// Get VMScaleSet Instance View
        /// List VMScaleSets in a RG
        /// List Available Skus
        /// Delete VMScaleSet
        /// Delete RG
        /// </summary>
        [Fact]
        [Trait("Name", "TestVMScaleSetScenarioOperations_Accept_Low")]
        public void TestVMScaleSetPriorityOperations_Accept_Low()
        {
            using (MockContext context = MockContext.Start(this.GetType()))
            {
                // Create low priority scaleset with no eviction policy specified. Eviction policy is defaulted to Deallocate.
                TestVMScaleSetPriorityOperationsInternal(context, VirtualMachinePriorityTypes.Low, hasManagedDisks: true);
            }
        }

        /// <summary>
        /// Covers following Operations:
        /// Create RG
        /// Create Storage Account
        /// Create Network Resources
        /// Create Azure Spot VMScaleSet with no EvictionPolicy specified
        /// Get VMScaleSet Model View
        /// Get VMScaleSet Instance View
        /// List VMScaleSets in a RG
        /// List Available Skus
        /// Delete VMScaleSet
        /// Delete RG
        /// </summary>
        [Fact]
        [Trait("Name", "TestVMScaleSetScenarioOperations_Accept_Spot")]
        public void TestVMScaleSetScenarioOperations_Accept_Spot()
        {
            using (MockContext context = MockContext.Start(this.GetType()))
            {
                // Create Azure Spot scaleset with no eviction policy specified. Eviction policy is defaulted to Deallocate.
                TestVMScaleSetPriorityOperationsInternal(context, VirtualMachinePriorityTypes.Spot, hasManagedDisks: true);
            }
        }

        /// <summary>
        /// Covers following Operations:
        /// Create RG
        /// Create Storage Account
        /// Create Network Resources
        /// Create Low Priority VMScaleSet with 'Delete' EvictionPolicy specified
        /// Delete VMScaleSet
        /// Delete RG
        /// </summary>
        [Fact]
        [Trait("Name", "TestVMScaleSetEvictionPolicyOperations_Accept_DeleteEvictionPolicy")]
        public void TestVMScaleSetEvictionPolicyOperations_Accept_DeleteEvictionPolicy()
        {
            using (MockContext context = MockContext.Start(this.GetType()))
            {
                // Create low priority scaleset with 'delete' eviction policy specified
                TestVMScaleSetPriorityOperationsInternal(context, VirtualMachinePriorityTypes.Low, evictionPolicy: VirtualMachineEvictionPolicyTypes.Delete, hasManagedDisks: true);
            }
        }

        #endregion

        #region Variable Pricing Tests

        /// <summary>
        /// Covers following Operations:
        /// Create RG
        /// Create Storage Account
        /// Create Network Resources
        /// Get VMScaleSet Model View
        /// Get VMScaleSet Instance View
        /// List VMScaleSets in a RG
        /// List Available Skus
        /// Delete VMScaleSet
        /// Delete RG
        /// </summary>
        [Fact]
        [Trait("Name", "TestVMScaleSetVariablePricedLowPriorityVM_Accept_DefaultMaxPrice")]
        public void TestVMScaleSetVariablePricedLowPriorityVM_Accept_DefaultMaxPrice()
        {
            using (MockContext context = MockContext.Start(this.GetType()))
            {
                TestVMScaleSetPriorityOperationsInternal(context, VirtualMachinePriorityTypes.Low, new BillingProfile { MaxPrice = -1 }, hasManagedDisks: true);
            }
        }

        /// <summary>
        /// Covers following Operations:
        /// Create RG
        /// Create Storage Account
        /// Create Network Resources
        /// Get VMScaleSet Model View
        /// Get VMScaleSet Instance View
        /// List VMScaleSets in a RG
        /// List Available Skus
        /// Delete VMScaleSet
        /// Delete RG
        /// </summary>
        [Fact]
        [Trait("Name", "TestVMScaleSetVariablePricedLowPriorityVM_Accept_UserSpecifiedMaxPrice")]
        public void TestVMScaleSetVariablePricedLowPriorityVM_Accept_UserSpecifiedMaxPrice()
        {
            using (MockContext context = MockContext.Start(this.GetType()))
            {
                TestVMScaleSetPriorityOperationsInternal(context, VirtualMachinePriorityTypes.Low, new BillingProfile { MaxPrice = 100 }, hasManagedDisks: true);
            }
        }

        #endregion

        private void TestVMScaleSetPriorityOperationsInternal(
            MockContext context,
            string priority,
            BillingProfile billingProfile = null,
            string evictionPolicy = null,
            bool hasManagedDisks = false,
            IList<string> zones = null
            )
        {
            string originalTestLocation = Environment.GetEnvironmentVariable("AZURE_VM_TEST_LOCATION");

            // Create resource group
            var rgName = TestUtilities.GenerateName(TestPrefix);
            var vmssName = TestUtilities.GenerateName("vmss");
            string storageAccountName = TestUtilities.GenerateName(TestPrefix);
            VirtualMachineScaleSet inputVMScaleSet;

            try
            {
                Environment.SetEnvironmentVariable("AZURE_VM_TEST_LOCATION", "centraluseuap");
                EnsureClientsInitialized(context);
                ImageReference imageRef = GetPlatformVMImage(useWindowsImage: true);

                var storageAccountOutput = CreateStorageAccount(rgName, storageAccountName);

                m_CrpClient.VirtualMachineScaleSets.Delete(rgName, "VMScaleSetDoesNotExist");

                var getResponse = CreateVMScaleSet_NoAsyncTracking(
                    rgName,
                    vmssName,
                    storageAccountOutput,
                    imageRef,
                    out inputVMScaleSet,
                    null,
                    (vmScaleSet) =>
                        {
                            vmScaleSet.Overprovision = true;
                            vmScaleSet.VirtualMachineProfile.Priority = priority;
                            vmScaleSet.VirtualMachineProfile.EvictionPolicy = evictionPolicy;
                            vmScaleSet.VirtualMachineProfile.BillingProfile = billingProfile;

                            vmScaleSet.Sku.Name = VirtualMachineSizeTypes.StandardA1;
                            vmScaleSet.Sku.Tier = "Standard";
                            vmScaleSet.Sku.Capacity = 2;
                        },
                    createWithManagedDisks: hasManagedDisks,
                    zones: zones);

                ValidateVMScaleSet(inputVMScaleSet, getResponse, hasManagedDisks);

                var getInstanceViewResponse = m_CrpClient.VirtualMachineScaleSets.GetInstanceView(rgName, vmssName);
                Assert.NotNull(getInstanceViewResponse);
                ValidateVMScaleSetInstanceView(inputVMScaleSet, getInstanceViewResponse);

                var listResponse = m_CrpClient.VirtualMachineScaleSets.List(rgName);
                ValidateVMScaleSet(
                    inputVMScaleSet,
                    listResponse.FirstOrDefault(x => x.Name == vmssName),
                    hasManagedDisks);

                var listSkusResponse = m_CrpClient.VirtualMachineScaleSets.ListSkus(rgName, vmssName);
                Assert.NotNull(listSkusResponse);
                Assert.False(listSkusResponse.Count() == 0);
                Assert.Same(inputVMScaleSet.VirtualMachineProfile.Priority.ToString(), priority);

                m_CrpClient.VirtualMachineScaleSets.Delete(rgName, vmssName);
            }
            finally
            {
                Environment.SetEnvironmentVariable("AZURE_VM_TEST_LOCATION", originalTestLocation);
                //Cleanup the created resources. But don't wait since it takes too long, and it's not the purpose
                //of the test to cover deletion. CSM does persistent retrying over all RG resources.
                m_ResourcesClient.ResourceGroups.Delete(rgName);
            }
        }
    }
}


text:
MEDUSA_BEGIN;

class OggAudio:public IAudio
{
public:
	using IAudio::IAudio;
public:
	virtual AudioFileType AudioType()const override { return AudioFileType::ogg; }

	static OggAudio* CreateFromFile(const FileIdRef& fileId);
	static OggAudio* CreateFromMemory(const FileIdRef& fileId,MemoryData data);

};

MEDUSA_END;
#endif


text:
'use strict';
const webpack = require('webpack');
const path = require('path');
const CircularDependencyPlugin = require('circular-dependency-plugin');

const debug = process.env.NODE_ENV !== 'production';
const productionPlugins = [
  new webpack.optimize.UglifyJsPlugin()
];

export default {
  devServer: {
    compress: true,
    port: 8002,
    historyApiFallback: true
  },
  entry: [
    path.join(__dirname, 'app', 'index.js')
  ],
  output: {
    path: path.join(__dirname, 'static', 'js'),
    filename: 'bundle.js',
    publicPath: '/js/'
  },
  resolve: {
    modules: [process.env.PWD, 'node_modules']
  },
  cache: false,
  devtool: debug ? 'source-map' : null,
  stats: {
    colors: true,
    reasons: true
  },
  module: {
    rules: [
      {
        test: /\.(js|jsx|graphql|gql)$/,
        exclude: /node_modules/,
        use: {
          loader: 'babel-loader',
          options: {
            cacheDirectory: true
          }
        }
      },
      {
        test: /\.(scss|sass)$/,
        use: [
          {
            loader: 'style-loader',
          },
          {
            loader: 'css-loader',
            options: {
              modules: true,
              localIdentName: debug ? '[local]__[hash:base64:5]' : '[hash:base64:5]'
            }
          },
          {
            loader: 'sass-loader',
            options: {
              outputStyle: 'expanded'
            }
          }
        ]
      },
      {
        test: /\.css$/,
        use: [
          {
            loader: 'style-loader',
          },
          {
            loader: 'css-loader',
            options: {
              modules: true
            }
          }
        ]
      },
      {
        test: /\.(png|jpg|woff|woff2|gif)$/,
        use: {
          loader: 'url-loader',
          options: {
            limit: 8192, prefix: '/'
          }
        }
      }
    ]
  },
  plugins: [
    new webpack.DefinePlugin({
      'process.env.NODE_ENV': JSON.stringify(process.env.NODE_ENV)
    }),
    new webpack.NoEmitOnErrorsPlugin(),
    new webpack.LoaderOptionsPlugin({
      debug: debug
    }),
    new webpack.optimize.CommonsChunkPlugin({
      children: true,
      async: true,
      minChunks: 2
    }),
    new CircularDependencyPlugin({
      exclude: 'node_modules',
      failOnError: true
    }),
    new webpack.HotModuleReplacementPlugin(),
    ...!debug && productionPlugins
  ]
}


text:
	<fieldset title="Thread 1">
		<legend>How to insert your camera?</legend>

Automatic Search

Manual Insert

Batch Insert

	</fieldset>
	<fieldset title="Thread 2">
		<legend>Manual Insert</legend>

				<input type="text" value="" name="modified" id="modified"/>

				Enable this camera
				<input type="checkbox" name="enable" id="enable" value='Yes' offval='No' checked="true"/>

				Camera name
				<input type="text" value="" name="name" id="name"/>

				IP address
				<input type="text" value="" name="address" id="address" class="required"/>

				Port
				<input type="text" value="80" name="port" id="port" size="11" maxlength="11"/>

				Camera brand
				<input id="brand" name="brand" />

				Camera model
				<input id="model" name="model" />

				Device channel<select id="view_channel" name="view_channel"></select>

				User name
				<input type="text" name="username" id="username" value="" />

				Password
				<input type="password" name="password" id="password" value="" />Press shift+enter to close virtual keyboard.

				Enable recordng on this camera
				<input type="checkbox" name="recording" id="recording" value='Yes' offval='No' />

	</fieldset>
	<fieldset title="Thread 3">
		<legend>Manual Insert</legend>
	</fieldset>
	<fieldset title="Thread 4">
		<legend>Manual Insert</legend>
	</fieldset>
	<fieldset title="Thread 5">
		<legend>Manual Insert</legend>
	</fieldset>


text:
{{--Nav bar--}}

        ChopBox

      @yield('navbar')

{{--Content--}}
@yield('content')


text:
package cat_osgi;

import org.osgi.framework.BundleActivator;
import org.osgi.framework.BundleContext;

public class Activator implements BundleActivator {

	/*
	 * (non-Javadoc)
	 * @see org.osgi.framework.BundleActivator#start(org.osgi.framework.BundleContext)
	 */
	public void start(BundleContext context) throws Exception {
		System.out.println("Hello World!!");
	}

	/*
	 * (non-Javadoc)
	 * @see org.osgi.framework.BundleActivator#stop(org.osgi.framework.BundleContext)
	 */
	public void stop(BundleContext context) throws Exception {
		System.out.println("Goodbye World!!");
	}

}


text:
from __future__ import (absolute_import, division, generators, nested_scopes, print_function,
                        unicode_literals, with_statement)

import ast

from pants.backend.python.tasks.checkstyle.common import CheckstylePlugin

class NewStyleClasses(CheckstylePlugin):
  """Enforce the use of new-style classes."""

  def nits(self):
    for class_def in self.iter_ast_types(ast.ClassDef):
      if not class_def.bases:
        yield self.error('T606', 'Classes must be new-style classes.', class_def)


text:
require "aws_agcod/version"
require "aws_agcod/config"
require "aws_agcod/create_gift_card"
require "aws_agcod/cancel_gift_card"
require "aws_agcod/gift_card_activity_list"

module AGCOD
  def self.configure(&block)
    @config = Config.new

    yield @config

    nil
  end

  def self.config
    @config
  end
end


text:
package net.voldrich.smscsim.spring.message;

import com.cloudhopper.smpp.pdu.PduRequest;

public interface MessageFactory<T extends PduRequest> {
	T createMessage() throws Exception;
}


text:
import {
  clickable,
  collection,
  is,
  interactor,
  Interactor,
  isPresent,
  isVisible,
  scoped,
  selectable,
} from '@bigtest/interactor';

import CalloutInteractor from '@folio/stripes-components/lib/Callout/tests/interactor'; // eslint-disable-line
import MultiSelectInteractor from '@folio/stripes-components/lib/MultiSelection/tests/interactor';
import RefundsReportModal from './refunds-report-modal';

@interactor class ActiveUserCheckbox {
  clickActive = clickable('#clickable-filter-active-active');
  clickInactive= clickable('#clickable-filter-active-inactive');
}

@interactor class DepartmentsFilter {
  multiSelect = new MultiSelectInteractor('#departments-filter');
  isAccordionPresent = isPresent('#users-filter-accordion-departments');
  whenLoaded() {
    return this.when(() => isPresent('#departments-filter'));
  }
}

@interactor class HeaderDropdown {
  click = clickable('[data-test-pane-header-actions-button]');
}

@interactor class HeaderDropdownMenu {
  clickExportToCSV = clickable('#export-overdue-loan-report');
  exportBtnIsVisible = isVisible('#export-overdue-loan-report');
  isExportBtnPresent = isPresent('#export-overdue-loan-report');
  isCashDrawerReportBtnPresent = isPresent('#cash-drawer-report');
  isFinancialTransactionReportBtnPresent = isPresent('#financial-transaction-report');
  clickRefundsReportCSV = clickable('#export-refunds-report');
  isRefundReportBtnVisible = isVisible('#export-refunds-report');
  isRefundsReportButtonPresent = isPresent('#export-refunds-report');
  columnCheckbox = function columnCheckbox(key) {
    return new Interactor(`[data-test-column-manager-checkbox="${key}"]`);
  }
}

@interactor class SearchFieldInteractor {
  static defaultScope = '[data-test-user-search-input]';
}

export default @interactor class UsersInteractor {
  static defaultScope = '[data-test-user-instances]';

  refundsReportModal = new RefundsReportModal();
  isRefundsReportModalPresent = isPresent('#refunds-report-modal');
  activeUserCheckbox = new ActiveUserCheckbox();
  departmentsFilter = new DepartmentsFilter();
  headerDropdownMenu = new HeaderDropdownMenu();
  searchField = new SearchFieldInteractor();
  searchButton = new Interactor('[data-test-user-search-submit]');
  searchFocused = isPresent('[data-test-user-search-input]:focus');
  chooseSearchOption = selectable('#input-user-search-qindex');
  paneHeaderFocused = is('#users-search-results-pane [class*=paneHeader---]', ':focus');
  patronGroupsPresent = isPresent('#clickable-filter-pg-faculty');
  instancePresent = isPresent('[data-test-instance-details]');
  instancesPresent = isPresent('[role=rowgroup] [role=row]');
  headerDropdown = new HeaderDropdown();
  clickFacultyCheckbox = clickable('#clickable-filter-pg-faculty');
  clickGraduateCheckbox = clickable('#clickable-filter-pg-graduate');
  clickStaffCheckbox = clickable('#clickable-filter-pg-staff');
  clickUndergradCheckbox = clickable('#clickable-filter-pg-undergrad');
  clickCreateUserButton = clickable('#clickable-newuser');
  instance = scoped('[data-test-instance-details]');
  instances = collection('[role=rowgroup] [data-row-inner]');
  clearStatusFilter = clickable('[data-test-clear-button]');
  column = function column(key) {
    return new Interactor(`#list-column-${key}`);
  }

  callout = new CalloutInteractor();

  whenInstanceLoaded() {
    return this.when(() => this.instancePresent).timeout(5000);
  }

  whenInstancesLoaded() {
    return this.when(() => this.instancesPresent).timeout(5000);
  }

  whenLoaded() {
    return this.when(() => this.patronGroupsPresent).timeout(5000);
  }
}


text:
#import <XXUnknownSuperclass.h> // Unknown library

@class NSString;

__attribute__((visibility("hidden")))
@interface Model_db_word030 : XXUnknownSuperclass {
	int _Index;
	int _WordId;
	NSString* _Options;
	NSString* _Answer;
	long _Version;
	NSString* _DataUId;
}
@property(retain, nonatomic) NSString* Answer;
@property(retain, nonatomic) NSString* DataUId;
@property(assign) int Index;
@property(retain, nonatomic) NSString* Options;
@property(assign, nonatomic) long Version;
@property(assign) int WordId;
- (id)init;
- (void).cxx_destruct;
@end


text:
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib
import assignment2_helper as helper

# Look pretty...
# matplotlib.style.use('ggplot')
plt.style.use('ggplot')

# Do * NOT * alter this line, until instructed!
scaleFeatures = False

# TODO: Load up the dataset and remove any and all
# Rows that have a nan. You should be a pro at this
# by now ;-)
#
# QUESTION: Should the id column be included as a
# feature?
#
# .. your code here ..

# Create some color coded labels; the actual label feature
# will be removed prior to executing PCA, since it's unsupervised.
# You're only labeling by color so you can see the effects of PCA
labels = ['red' if i=='ckd' else 'green' for i in df.classification]

# TODO: Use an indexer to select only the following columns:
#       ['bgr','wc','rc']
#
# .. your code here ..

# TODO: Print out and check your dataframe's dtypes. You'll might
# want to set a breakpoint after you print it out so you can stop the
# program's execution.
#
# You can either take a look at the dataset webpage in the attribute info
# section: https://archive.ics.uci.edu/ml/datasets/Chronic_Kidney_Disease
# or you can actually peek through the dataframe by printing a few rows.
# What kind of data type should these three columns be? If Pandas didn't
# properly detect and convert them to that data type for you, then use
# an appropriate command to coerce these features into the right type.
#
# .. your code here ..

# TODO: PCA Operates based on variance. The variable with the greatest
# variance will dominate. Go ahead and peek into your data using a
# command that will check the variance of every feature in your dataset.
# Print out the results. Also print out the results of running .describe
# on your dataset.
#
# Hint: If you don't see all three variables: 'bgr','wc' and 'rc', then
# you probably didn't complete the previous step properly.
#
# .. your code here ..

# TODO: This method assumes your dataframe is called df. If it isn't,
# make the appropriate changes. Don't alter the code in scaleFeatures()
# just yet though!
#
# .. your code adjustment here ..
if scaleFeatures: df = helper.scaleFeatures(df)

# TODO: Run PCA on your dataset and reduce it to 2 components
# Ensure your PCA instance is saved in a variable called 'pca',
# and that the results of your transformation are saved in 'T'.
#
# .. your code here ..

# Plot the transformed data as a scatter plot. Recall that transforming
# the data will result in a NumPy NDArray. You can either use MatPlotLib
# to graph it directly, or you can convert it to DataFrame and have pandas
# do it for you.
#
# Since we've already demonstrated how to plot directly with MatPlotLib in
# Module4/assignment1.py, this time we'll convert to a Pandas Dataframe.
#
# Since we transformed via PCA, we no longer have column names. We know we
# are in P.C. space, so we'll just define the coordinates accordingly:
ax = helper.drawVectors(T, pca.components_, df.columns.values, plt, scaleFeatures)
T = pd.DataFrame(T)
T.columns = ['component1', 'component2']
T.plot.scatter(x='component1', y='component2', marker='o', c=labels, alpha=0.75, ax=ax)
plt.show()


text:
div {
  background-image: url("prepend/relative.jpg");
}

div {
  background-image: url("prepend/absolute.jpg");
}

div {
  background-image: url("data:image/jpg;base64,/9j/ke4uvs");
}


text:
package net.idlesoft.android.apps.github.ui.adapters;

import net.idlesoft.android.apps.github.R;
import net.idlesoft.android.apps.github.ui.activities.BaseActivity;

import android.view.View;
import android.view.ViewGroup;
import android.widget.AdapterView;
import android.widget.TextView;

public class InfoListAdapter extends BaseListAdapter<InfoListAdapter.InfoHolder> {

    public static class InfoHolder {

        public
        String primary;

        public
        String secondary;

        public
        AdapterView.OnItemClickListener onClick;

        public
        AdapterView.OnItemLongClickListener onLongClick;
    }

    public static class ViewHolder {

        TextView primary;

        TextView secondary;
    }

    public InfoListAdapter(BaseActivity context) {
        super(context);
    }

    @Override
    public View getView(int position, View convertView, ViewGroup parent) {
        ViewHolder viewHolder;

        if (convertView == null) {
            convertView = mInflater.inflate(R.layout.info_list_item, null);
            viewHolder = new ViewHolder();
            viewHolder.primary = (TextView) convertView.findViewById(R.id.tv_info_primary);
            viewHolder.secondary = (TextView) convertView.findViewById(R.id.tv_info_secondary);
            convertView.setTag(viewHolder);
        } else {
            viewHolder = (ViewHolder) convertView.getTag();
        }

        final InfoHolder holder = getItem(position);

        viewHolder.primary.setText(holder.primary);
        viewHolder.secondary.setText(holder.secondary);

        return convertView;
    }

}


text:
type WebpackMap = {
  [filepath: string]: {
    [name: string]: ModuleMetaData,
  },
};

export type BundlerConfig = WebpackMap;

// eslint-disable-next-line no-unused-vars
export type ModuleReference<T> = {
  $$typeof: Symbol,
  filepath: string,
  name: string,
};

export type ModuleMetaData = {
  id: string,
  chunks: Array<string>,
  name: string,
};

export type ModuleKey = string;

const MODULE_TAG = Symbol.for('react.module.reference');

export function getModuleKey(reference: ModuleReference<any>): ModuleKey {
  return reference.filepath + '#' + reference.name;
}

export function isModuleReference(reference: Object): boolean {
  return reference.$$typeof === MODULE_TAG;
}

export function resolveModuleMetaData<T>(
  config: BundlerConfig,
  moduleReference: ModuleReference<T>,
): ModuleMetaData {
  return config[moduleReference.filepath][moduleReference.name];
}


text:
module RedisFailover
  # Provides manual failover support to a new master.
  class ManualFailover
    # Path for manual failover communication.
    ZNODE_PATH = 'manual_failover'.freeze

    # Denotes that any slave can be used as a candidate for promotion.
    ANY_SLAVE = "ANY_SLAVE".freeze

    def self.path(root_znode)
      "#{root_znode}/#{ZNODE_PATH}"
    end

    # Creates a new instance.
    #
    # @param [ZK] zk the ZooKeeper client
    # @param [ZNode] root_znode the root ZK node
    # @param [Hash] options the options used for manual failover
    # @option options [String] :host the host of the failover candidate
    # @option options [String] :port the port of the failover candidate
    # @note
    #   If options is empty, a random slave will be used
    #   as a failover candidate.
    def initialize(zk, root_znode, options = {})
      @zk = zk
      @root_znode = root_znode
      @options = options

      unless @options.empty?
        port = Integer(@options[:port]) rescue nil
        raise ArgumentError, ':host not properly specified' if @options[:host].to_s.empty?
        raise ArgumentError, ':port not properly specified' if port.nil?
      end
    end

    # Performs a manual failover.
    def perform
      create_path
      node = @options.empty? ? ANY_SLAVE : "#{@options[:host]}:#{@options[:port]}"
      @zk.set(self.class.path(@root_znode), node)
    end

    private

    # Creates the znode path used for coordinating manual failovers.
    def create_path
      @zk.create(self.class.path(@root_znode))
    rescue ZK::Exceptions::NodeExists
      # best effort
    end
  end
end


text:
Jon Surrell

[jonsurrell.com](http://jonsurrell.com)

Desarrollo en [Click Army](https://www.clickarmy.net/)

Auto-didacta

Juego a Ultimate Frisbee

Llevo Crocs®

Note:
Crocs!
Funcional.


text:
Parallel Cartographic Modeling Language - PCML
==============================================

Introduction
------------

The parallel cartographic modeling language (PCML) is a multi-institutional
collaborative project aiming to create a computing language for
cyberGIScientists that is designed for (1) usability, (2) programmability, and
(3) scalability. PCML provides multi-core parallel processing for spatial
operations while hiding the implementation complexities of parallelism.

Installation
------------

### 1. Make sure that `pip` and `setuptools` are installed in your current Python environment (global or a virtualenv).

### 2. Install GDAL library

    $ sudo apt-get install libgdal-dev

or

    $ su -c 'yum install gdal-devel gdal-libs'

*The command may vary according to package manager and system.*

### 3. Install the required dependencies

    $ pip install -r requirements.txt

### 4. Finally, install

    $ python setup.py install

<!-- TODO: platform/distribution specific troubleshooting. -->

Example
-------

    from pcml import *
    PCMLConfig.num_procs = 4 # Run computation in 4 processes (default)
    layer1 = ReadASCIIGrid("layer1.asc")
    layer2 = ReadGeoTIFF("layer2.tiff")
    layer_out = layer1 + layer2
    layer_out.print_data()

Please also see `test.py` for additional working examples.

Build status
------------

[![wercker status](https://app.wercker.com/status/99dd16339b190c2ab04db505fa7af57a/m "wercker status")](https://app.wercker.com/project/bykey/99dd16339b190c2ab04db505fa7af57a)


text:
class CongressionalMapper < ActiveRecord::Base
	belongs_to :state
end


text:
import argparse
import logging
import os
import sys

if __name__ == '__main__':
  sys.path.append(os.path.abspath(os.path.join(
      os.path.dirname(__file__), '..', '..', '..')))
from devil.android import device_blacklist
from devil.android import device_utils
from devil.android import fastboot_utils
from devil.android.tools import script_common
from devil.constants import exit_codes
from devil.utils import run_tests_helper

def main():
  parser = argparse.ArgumentParser()
  parser.add_argument('build_path', help='Path to android build.')
  parser.add_argument('-d', '--device', dest='devices', action='append',
                      help='Device(s) to flash.')
  parser.add_argument('-v', '--verbose', default=0, action='count',
                      help='Verbose level (multiple times for more)')
  parser.add_argument('-w', '--wipe', action='store_true',
                       help='If set, wipes user data')
  parser.add_argument('--blacklist-file', help='Device blacklist file.')
  args = parser.parse_args()
  run_tests_helper.SetLogLevel(args.verbose)

  if args.blacklist_file:
    blacklist = device_blacklist.Blacklist(args.blacklist_file).Read()
    if blacklist:
      logging.critical('Device(s) in blacklist, not flashing devices:')
      for key in blacklist:
        logging.critical('  %s', key)
      return exit_codes.INFRA

  flashed_devices = []
  failed_devices = []

  def flash(device):
    fastboot = fastboot_utils.FastbootUtils(device)
    try:
      fastboot.FlashDevice(args.build_path, wipe=args.wipe)
      flashed_devices.append(device)
    except Exception:  # pylint: disable=broad-except
      logging.exception('Device %s failed to flash.', str(device))
      failed_devices.append(device)

  devices = script_common.GetDevices(args.devices, args.blacklist_file)
  device_utils.DeviceUtils.parallel(devices).pMap(flash)

  if flashed_devices:
    logging.info('The following devices were flashed:')
    logging.info('  %s', ' '.join(str(d) for d in flashed_devices))
  if failed_devices:
    logging.critical('The following devices failed to flash:')
    logging.critical('  %s', ' '.join(str(d) for d in failed_devices))
    return exit_codes.INFRA
  return 0

if __name__ == '__main__':
  sys.exit(main())


text:
package aws

import (
	"log"

	"encoding/json"
	"fmt"
	"io/ioutil"
	"net/http"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/service/emr"
	"github.com/hashicorp/terraform/helper/resource"
	"github.com/hashicorp/terraform/helper/schema"
)

func resourceAwsEMRCluster() *schema.Resource {
	return &schema.Resource{
		Create: resourceAwsEMRClusterCreate,
		Read:   resourceAwsEMRClusterRead,
		Update: resourceAwsEMRClusterUpdate,
		Delete: resourceAwsEMRClusterDelete,
		Schema: map[string]*schema.Schema{
			"name": {
				Type:     schema.TypeString,
				ForceNew: true,
				Required: true,
			},
			"release_label": {
				Type:     schema.TypeString,
				ForceNew: true,
				Required: true,
			},
			"master_instance_type": {
				Type:     schema.TypeString,
				Required: false,
				Optional: true,
				ForceNew: true,
			},
			"core_instance_type": {
				Type:     schema.TypeString,
				Optional: true,
				ForceNew: true,
				Computed: true,
			},
			"core_instance_count": {
				Type:     schema.TypeInt,
				Optional: true,
			},
			"cluster_state": {
				Type:     schema.TypeString,
				Computed: true,
			},
			"log_uri": {
				Type:     schema.TypeString,
				ForceNew: true,
				Optional: true,
				DiffSuppressFunc: func(k, old, new string, d *schema.ResourceData) bool {
					// EMR uses a proprietary filesystem called EMRFS
					// and both s3n & s3 protocols are mapped to that FS
					// so they're equvivalent in this context (confirmed by AWS support)
					old = strings.Replace(old, "s3n://", "s3://", -1)
					return old == new
				},
			},
			"master_public_dns": {
				Type:     schema.TypeString,
				Computed: true,
			},
			"applications": {
				Type:     schema.TypeSet,
				Optional: true,
				ForceNew: true,
				Elem:     &schema.Schema{Type: schema.TypeString},
				Set:      schema.HashString,
			},
			"termination_protection": {
				Type:     schema.TypeBool,
				Optional: true,
				Computed: true,
			},
			"keep_job_flow_alive_when_no_steps": {
				Type:     schema.TypeBool,
				ForceNew: true,
				Optional: true,
				Computed: true,
			},
			"ec2_attributes": {
				Type:     schema.TypeList,
				MaxItems: 1,
				Optional: true,
				ForceNew: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"key_name": {
							Type:     schema.TypeString,
							Optional: true,
						},
						"subnet_id": {
							Type:     schema.TypeString,
							Optional: true,
						},
						"additional_master_security_groups": {
							Type:     schema.TypeString,
							Optional: true,
						},
						"additional_slave_security_groups": {
							Type:     schema.TypeString,
							Optional: true,
						},
						"emr_managed_master_security_group": {
							Type:     schema.TypeString,
							Optional: true,
						},
						"emr_managed_slave_security_group": {
							Type:     schema.TypeString,
							Optional: true,
						},
						"instance_profile": {
							Type:     schema.TypeString,
							Required: true,
						},
						"service_access_security_group": {
							Type:     schema.TypeString,
							Optional: true,
						},
					},
				},
			},
			"instance_group": {
				Type:     schema.TypeSet,
				Optional: true,
				ForceNew: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"bid_price": {
							Type:     schema.TypeString,
							Optional: true,
							Required: false,
						},
						"ebs_config": {
							Type:     schema.TypeSet,
							Optional: true,
							ForceNew: true,
							Elem: &schema.Resource{
								Schema: map[string]*schema.Schema{
									"iops": {
										Type:     schema.TypeInt,
										Optional: true,
									},
									"size": {
										Type:     schema.TypeInt,
										Required: true,
									},
									"type": {
										Type:         schema.TypeString,
										Required:     true,
										ValidateFunc: validateAwsEmrEbsVolumeType,
									},
									"volumes_per_instance": {
										Type:     schema.TypeInt,
										Optional: true,
										Default:  1,
									},
								},
							},
						},
						"instance_count": {
							Type:     schema.TypeInt,
							Optional: true,
							Default:  0,
						},
						"instance_role": {
							Type:         schema.TypeString,
							Required:     true,
							ValidateFunc: validateAwsEmrInstanceGroupRole,
						},
						"instance_type": {
							Type:     schema.TypeString,
							Required: true,
							ForceNew: true,
						},
						"name": {
							Type:     schema.TypeString,
							Optional: true,
							ForceNew: true,
						},
					},
				},
			},
			"bootstrap_action": {
				Type:     schema.TypeSet,
				Optional: true,
				ForceNew: true,
				Elem: &schema.Resource{
					Schema: map[string]*schema.Schema{
						"name": {
							Type:     schema.TypeString,
							Required: true,
						},
						"path": {
							Type:     schema.TypeString,
							Required: true,
						},
						"args": {
							Type:     schema.TypeList,
							Optional: true,
							ForceNew: true,
							Elem:     &schema.Schema{Type: schema.TypeString},
						},
					},
				},
			},
			"tags": tagsSchema(),
			"configurations": {
				Type:     schema.TypeString,
				ForceNew: true,
				Optional: true,
			},
			"service_role": {
				Type:     schema.TypeString,
				ForceNew: true,
				Required: true,
			},
			"security_configuration": {
				Type:     schema.TypeString,
				ForceNew: true,
				Optional: true,
			},
			"autoscaling_role": &schema.Schema{
				Type:     schema.TypeString,
				ForceNew: true,
				Optional: true,
			},
			"visible_to_all_users": {
				Type:     schema.TypeBool,
				Optional: true,
				Default:  true,
			},
			"ebs_root_volume_size": {
				Type:     schema.TypeInt,
				ForceNew: true,
				Optional: true,
			},
		},
	}
}

func resourceAwsEMRClusterCreate(d *schema.ResourceData, meta interface{}) error {
	conn := meta.(*AWSClient).emrconn

	log.Printf("[DEBUG] Creating EMR cluster")
	applications := d.Get("applications").(*schema.Set).List()

	keepJobFlowAliveWhenNoSteps := true
	if v, ok := d.GetOk("keep_job_flow_alive_when_no_steps"); ok {
		keepJobFlowAliveWhenNoSteps = v.(bool)
	}

	terminationProtection := false
	if v, ok := d.GetOk("termination_protection"); ok {
		terminationProtection = v.(bool)
	}
	instanceConfig := &emr.JobFlowInstancesConfig{
		KeepJobFlowAliveWhenNoSteps: aws.Bool(keepJobFlowAliveWhenNoSteps),
		TerminationProtected:        aws.Bool(terminationProtection),
	}

	if v, ok := d.GetOk("master_instance_type"); ok {
		instanceConfig.MasterInstanceType = aws.String(v.(string))
		instanceConfig.SlaveInstanceType = aws.String(v.(string))
	}
	if v, ok := d.GetOk("core_instance_type"); ok {
		instanceConfig.SlaveInstanceType = aws.String(v.(string))
	}
	if v, ok := d.GetOk("core_instance_count"); ok {
		instanceConfig.InstanceCount = aws.Int64(int64(v.(int)))
	}

	var instanceProfile string
	if a, ok := d.GetOk("ec2_attributes"); ok {
		ec2Attributes := a.([]interface{})
		attributes := ec2Attributes[0].(map[string]interface{})

		if v, ok := attributes["key_name"]; ok {
			instanceConfig.Ec2KeyName = aws.String(v.(string))
		}
		if v, ok := attributes["subnet_id"]; ok {
			instanceConfig.Ec2SubnetId = aws.String(v.(string))
		}
		if v, ok := attributes["subnet_id"]; ok {
			instanceConfig.Ec2SubnetId = aws.String(v.(string))
		}

		if v, ok := attributes["additional_master_security_groups"]; ok {
			strSlice := strings.Split(v.(string), ",")
			for i, s := range strSlice {
				strSlice[i] = strings.TrimSpace(s)
			}
			instanceConfig.AdditionalMasterSecurityGroups = aws.StringSlice(strSlice)
		}

		if v, ok := attributes["additional_slave_security_groups"]; ok {
			strSlice := strings.Split(v.(string), ",")
			for i, s := range strSlice {
				strSlice[i] = strings.TrimSpace(s)
			}
			instanceConfig.AdditionalSlaveSecurityGroups = aws.StringSlice(strSlice)
		}

		if v, ok := attributes["emr_managed_master_security_group"]; ok {
			instanceConfig.EmrManagedMasterSecurityGroup = aws.String(v.(string))
		}
		if v, ok := attributes["emr_managed_slave_security_group"]; ok {
			instanceConfig.EmrManagedSlaveSecurityGroup = aws.String(v.(string))
		}

		if len(strings.TrimSpace(attributes["instance_profile"].(string))) != 0 {
			instanceProfile = strings.TrimSpace(attributes["instance_profile"].(string))
		}

		if v, ok := attributes["service_access_security_group"]; ok {
			instanceConfig.ServiceAccessSecurityGroup = aws.String(v.(string))
		}
	}
	if v, ok := d.GetOk("instance_group"); ok {
		instanceGroupConfigs := v.(*schema.Set).List()
		instanceConfig.InstanceGroups = expandInstanceGroupConfigs(instanceGroupConfigs)
	}

	emrApps := expandApplications(applications)

	params := &emr.RunJobFlowInput{
		Instances:    instanceConfig,
		Name:         aws.String(d.Get("name").(string)),
		Applications: emrApps,

		ReleaseLabel:      aws.String(d.Get("release_label").(string)),
		ServiceRole:       aws.String(d.Get("service_role").(string)),
		VisibleToAllUsers: aws.Bool(d.Get("visible_to_all_users").(bool)),
	}

	if v, ok := d.GetOk("log_uri"); ok {
		params.LogUri = aws.String(v.(string))
	}
	if v, ok := d.GetOk("autoscaling_role"); ok {
		params.AutoScalingRole = aws.String(v.(string))
	}

	if v, ok := d.GetOk("security_configuration"); ok {
		params.SecurityConfiguration = aws.String(v.(string))
	}

	if v, ok := d.GetOk("ebs_root_volume_size"); ok {
		params.EbsRootVolumeSize = aws.Int64(int64(v.(int)))
	}

	if instanceProfile != "" {
		params.JobFlowRole = aws.String(instanceProfile)
	}

	if v, ok := d.GetOk("bootstrap_action"); ok {
		bootstrapActions := v.(*schema.Set).List()
		params.BootstrapActions = expandBootstrapActions(bootstrapActions)
	}
	if v, ok := d.GetOk("tags"); ok {
		tagsIn := v.(map[string]interface{})
		params.Tags = expandTags(tagsIn)
	}
	if v, ok := d.GetOk("configurations"); ok {
		confUrl := v.(string)
		params.Configurations = expandConfigures(confUrl)
	}

	log.Printf("[DEBUG] EMR Cluster create options: %s", params)

	var resp *emr.RunJobFlowOutput
	err := resource.Retry(30*time.Second, func() *resource.RetryError {
		var err error
		resp, err = conn.RunJobFlow(params)
		if err != nil {
			if isAWSErr(err, "ValidationException", "Invalid InstanceProfile:") {
				return resource.RetryableError(err)
			}
			if isAWSErr(err, "AccessDeniedException", "Failed to authorize instance profile") {
				return resource.RetryableError(err)
			}
			return resource.NonRetryableError(err)
		}
		return nil
	})
	if err != nil {
		return err
	}

	d.SetId(*resp.JobFlowId)

	log.Println("[INFO] Waiting for EMR Cluster to be available")

	stateConf := &resource.StateChangeConf{
		Pending:    []string{"STARTING", "BOOTSTRAPPING"},
		Target:     []string{"WAITING", "RUNNING"},
		Refresh:    resourceAwsEMRClusterStateRefreshFunc(d, meta),
		Timeout:    75 * time.Minute,
		MinTimeout: 10 * time.Second,
		Delay:      30 * time.Second, // Wait 30 secs before starting
	}

	_, err = stateConf.WaitForState()
	if err != nil {
		return fmt.Errorf("[WARN] Error waiting for EMR Cluster state to be \"WAITING\" or \"RUNNING\": %s", err)
	}

	return resourceAwsEMRClusterRead(d, meta)
}

func resourceAwsEMRClusterRead(d *schema.ResourceData, meta interface{}) error {
	emrconn := meta.(*AWSClient).emrconn

	req := &emr.DescribeClusterInput{
		ClusterId: aws.String(d.Id()),
	}

	resp, err := emrconn.DescribeCluster(req)
	if err != nil {
		return fmt.Errorf("Error reading EMR cluster: %s", err)
	}

	if resp.Cluster == nil {
		log.Printf("[DEBUG] EMR Cluster (%s) not found", d.Id())
		d.SetId("")
		return nil
	}

	cluster := resp.Cluster

	if cluster.Status != nil {
		if *cluster.Status.State == "TERMINATED" {
			log.Printf("[DEBUG] EMR Cluster (%s) was TERMINATED already", d.Id())
			d.SetId("")
			return nil
		}

		if *cluster.Status.State == "TERMINATED_WITH_ERRORS" {
			log.Printf("[DEBUG] EMR Cluster (%s) was TERMINATED_WITH_ERRORS already", d.Id())
			d.SetId("")
			return nil
		}

		d.Set("cluster_state", cluster.Status.State)
	}

	instanceGroups, err := fetchAllEMRInstanceGroups(emrconn, d.Id())
	if err == nil {
		coreGroup := findGroup(instanceGroups, "CORE")
		if coreGroup != nil {
			d.Set("core_instance_type", coreGroup.InstanceType)
		}
		if err := d.Set("instance_group", flattenInstanceGroups(instanceGroups)); err != nil {
			log.Printf("[ERR] Error setting EMR instance groups: %s", err)
		}
	}

	d.Set("name", cluster.Name)
	d.Set("service_role", cluster.ServiceRole)
	d.Set("security_configuration", cluster.SecurityConfiguration)
	d.Set("autoscaling_role", cluster.AutoScalingRole)
	d.Set("release_label", cluster.ReleaseLabel)
	d.Set("log_uri", cluster.LogUri)
	d.Set("master_public_dns", cluster.MasterPublicDnsName)
	d.Set("visible_to_all_users", cluster.VisibleToAllUsers)
	d.Set("tags", tagsToMapEMR(cluster.Tags))
	d.Set("ebs_root_volume_size", cluster.EbsRootVolumeSize)

	if err := d.Set("applications", flattenApplications(cluster.Applications)); err != nil {
		log.Printf("[ERR] Error setting EMR Applications for cluster (%s): %s", d.Id(), err)
	}

	// Configurations is a JSON document. It's built with an expand method but a
	// simple string should be returned as JSON
	if err := d.Set("configurations", cluster.Configurations); err != nil {
		log.Printf("[ERR] Error setting EMR configurations for cluster (%s): %s", d.Id(), err)
	}

	if err := d.Set("ec2_attributes", flattenEc2Attributes(cluster.Ec2InstanceAttributes)); err != nil {
		log.Printf("[ERR] Error setting EMR Ec2 Attributes: %s", err)
	}

	respBootstraps, err := emrconn.ListBootstrapActions(&emr.ListBootstrapActionsInput{
		ClusterId: cluster.Id,
	})
	if err != nil {
		log.Printf("[WARN] Error listing bootstrap actions: %s", err)
	}

	if err := d.Set("bootstrap_action", flattenBootstrapArguments(respBootstraps.BootstrapActions)); err != nil {
		log.Printf("[WARN] Error setting Bootstrap Actions: %s", err)
	}

	return nil
}

func resourceAwsEMRClusterUpdate(d *schema.ResourceData, meta interface{}) error {
	conn := meta.(*AWSClient).emrconn

	d.Partial(true)

	if d.HasChange("core_instance_count") {
		d.SetPartial("core_instance_count")
		log.Printf("[DEBUG] Modify EMR cluster")
		groups, err := fetchAllEMRInstanceGroups(conn, d.Id())
		if err != nil {
			log.Printf("[DEBUG] Error finding all instance groups: %s", err)
			return err
		}

		coreInstanceCount := d.Get("core_instance_count").(int)
		coreGroup := findGroup(groups, "CORE")
		if coreGroup == nil {
			return fmt.Errorf("[ERR] Error finding core group")
		}

		params := &emr.ModifyInstanceGroupsInput{
			InstanceGroups: []*emr.InstanceGroupModifyConfig{
				{
					InstanceGroupId: coreGroup.Id,
					InstanceCount:   aws.Int64(int64(coreInstanceCount) - 1),
				},
			},
		}
		_, errModify := conn.ModifyInstanceGroups(params)
		if errModify != nil {
			log.Printf("[ERROR] %s", errModify)
			return errModify
		}

		log.Printf("[DEBUG] Modify EMR Cluster done...")

		log.Println("[INFO] Waiting for EMR Cluster to be available")

		stateConf := &resource.StateChangeConf{
			Pending:    []string{"STARTING", "BOOTSTRAPPING"},
			Target:     []string{"WAITING", "RUNNING"},
			Refresh:    resourceAwsEMRClusterStateRefreshFunc(d, meta),
			Timeout:    40 * time.Minute,
			MinTimeout: 10 * time.Second,
			Delay:      5 * time.Second,
		}

		_, err = stateConf.WaitForState()
		if err != nil {
			return fmt.Errorf("[WARN] Error waiting for EMR Cluster state to be \"WAITING\" or \"RUNNING\" after modification: %s", err)
		}
	}

	if d.HasChange("visible_to_all_users") {
		d.SetPartial("visible_to_all_users")
		_, errModify := conn.SetVisibleToAllUsers(&emr.SetVisibleToAllUsersInput{
			JobFlowIds:        []*string{aws.String(d.Id())},
			VisibleToAllUsers: aws.Bool(d.Get("visible_to_all_users").(bool)),
		})
		if errModify != nil {
			log.Printf("[ERROR] %s", errModify)
			return errModify
		}
	}

	if d.HasChange("termination_protection") {
		d.SetPartial("termination_protection")
		_, errModify := conn.SetTerminationProtection(&emr.SetTerminationProtectionInput{
			JobFlowIds:           []*string{aws.String(d.Id())},
			TerminationProtected: aws.Bool(d.Get("termination_protection").(bool)),
		})
		if errModify != nil {
			log.Printf("[ERROR] %s", errModify)
			return errModify
		}
	}

	if err := setTagsEMR(conn, d); err != nil {
		return err
	} else {
		d.SetPartial("tags")
	}

	d.Partial(false)

	return resourceAwsEMRClusterRead(d, meta)
}

func resourceAwsEMRClusterDelete(d *schema.ResourceData, meta interface{}) error {
	conn := meta.(*AWSClient).emrconn

	req := &emr.TerminateJobFlowsInput{
		JobFlowIds: []*string{
			aws.String(d.Id()),
		},
	}

	_, err := conn.TerminateJobFlows(req)
	if err != nil {
		log.Printf("[ERROR], %s", err)
		return err
	}

	err = resource.Retry(10*time.Minute, func() *resource.RetryError {
		resp, err := conn.ListInstances(&emr.ListInstancesInput{
			ClusterId: aws.String(d.Id()),
		})

		if err != nil {
			return resource.NonRetryableError(err)
		}

		instanceCount := len(resp.Instances)

		if resp == nil || instanceCount == 0 {
			log.Printf("[DEBUG] No instances found for EMR Cluster (%s)", d.Id())
			return nil
		}

		// Collect instance status states, wait for all instances to be terminated
		// before moving on
		var terminated []string
		for j, i := range resp.Instances {
			if i.Status != nil {
				if *i.Status.State == "TERMINATED" {
					terminated = append(terminated, *i.Ec2InstanceId)
				}
			} else {
				log.Printf("[DEBUG] Cluster instance (%d : %s) has no status", j, *i.Ec2InstanceId)
			}
		}
		if len(terminated) == instanceCount {
			log.Printf("[DEBUG] All (%d) EMR Cluster (%s) Instances terminated", instanceCount, d.Id())
			return nil
		}
		return resource.RetryableError(fmt.Errorf("[DEBUG] EMR Cluster (%s) has (%d) Instances remaining, retrying", d.Id(), len(resp.Instances)))
	})

	if err != nil {
		log.Printf("[ERR] Error waiting for EMR Cluster (%s) Instances to drain", d.Id())
	}

	d.SetId("")
	return nil
}

func expandApplications(apps []interface{}) []*emr.Application {
	appOut := make([]*emr.Application, 0, len(apps))

	for _, appName := range expandStringList(apps) {
		app := &emr.Application{
			Name: appName,
		}
		appOut = append(appOut, app)
	}
	return appOut
}

func flattenApplications(apps []*emr.Application) []interface{} {
	appOut := make([]interface{}, 0, len(apps))

	for _, app := range apps {
		appOut = append(appOut, *app.Name)
	}
	return appOut
}

func flattenEc2Attributes(ia *emr.Ec2InstanceAttributes) []map[string]interface{} {
	attrs := map[string]interface{}{}
	result := make([]map[string]interface{}, 0)

	if ia.Ec2KeyName != nil {
		attrs["key_name"] = *ia.Ec2KeyName
	}
	if ia.Ec2SubnetId != nil {
		attrs["subnet_id"] = *ia.Ec2SubnetId
	}
	if ia.IamInstanceProfile != nil {
		attrs["instance_profile"] = *ia.IamInstanceProfile
	}
	if ia.EmrManagedMasterSecurityGroup != nil {
		attrs["emr_managed_master_security_group"] = *ia.EmrManagedMasterSecurityGroup
	}
	if ia.EmrManagedSlaveSecurityGroup != nil {
		attrs["emr_managed_slave_security_group"] = *ia.EmrManagedSlaveSecurityGroup
	}

	if len(ia.AdditionalMasterSecurityGroups) > 0 {
		strs := aws.StringValueSlice(ia.AdditionalMasterSecurityGroups)
		attrs["additional_master_security_groups"] = strings.Join(strs, ",")
	}
	if len(ia.AdditionalSlaveSecurityGroups) > 0 {
		strs := aws.StringValueSlice(ia.AdditionalSlaveSecurityGroups)
		attrs["additional_slave_security_groups"] = strings.Join(strs, ",")
	}

	if ia.ServiceAccessSecurityGroup != nil {
		attrs["service_access_security_group"] = *ia.ServiceAccessSecurityGroup
	}

	result = append(result, attrs)

	return result
}

func flattenInstanceGroups(igs []*emr.InstanceGroup) []map[string]interface{} {
	result := make([]map[string]interface{}, 0)

	for _, ig := range igs {
		attrs := make(map[string]interface{})
		if ig.BidPrice != nil {
			attrs["bid_price"] = *ig.BidPrice
		} else {
			attrs["bid_price"] = ""
		}
		ebsConfig := make([]map[string]interface{}, 0)
		for _, ebs := range ig.EbsBlockDevices {
			ebsAttrs := make(map[string]interface{})
			if ebs.VolumeSpecification.Iops != nil {
				ebsAttrs["iops"] = *ebs.VolumeSpecification.Iops
			} else {
				ebsAttrs["iops"] = ""
			}
			ebsAttrs["size"] = *ebs.VolumeSpecification.SizeInGB
			ebsAttrs["type"] = *ebs.VolumeSpecification.VolumeType
			ebsAttrs["volumes_per_instance"] = 1

			ebsConfig = append(ebsConfig, ebsAttrs)
		}
		attrs["ebs_config"] = ebsConfig
		attrs["instance_count"] = *ig.RequestedInstanceCount
		attrs["instance_role"] = *ig.InstanceGroupType
		attrs["instance_type"] = *ig.InstanceType
		attrs["name"] = *ig.Name
		result = append(result, attrs)
	}

	return result
}

func flattenBootstrapArguments(actions []*emr.Command) []map[string]interface{} {
	result := make([]map[string]interface{}, 0)

	for _, b := range actions {
		attrs := make(map[string]interface{})
		attrs["name"] = *b.Name
		attrs["path"] = *b.ScriptPath
		attrs["args"] = flattenStringList(b.Args)
		result = append(result, attrs)
	}

	return result
}

func loadGroups(d *schema.ResourceData, meta interface{}) ([]*emr.InstanceGroup, error) {
	emrconn := meta.(*AWSClient).emrconn
	reqGrps := &emr.ListInstanceGroupsInput{
		ClusterId: aws.String(d.Id()),
	}

	respGrps, errGrps := emrconn.ListInstanceGroups(reqGrps)
	if errGrps != nil {
		return nil, fmt.Errorf("Error reading EMR cluster: %s", errGrps)
	}
	return respGrps.InstanceGroups, nil
}

func findGroup(grps []*emr.InstanceGroup, typ string) *emr.InstanceGroup {
	for _, grp := range grps {
		if grp.InstanceGroupType != nil {
			if *grp.InstanceGroupType == typ {
				return grp
			}
		}
	}
	return nil
}

func expandTags(m map[string]interface{}) []*emr.Tag {
	var result []*emr.Tag
	for k, v := range m {
		result = append(result, &emr.Tag{
			Key:   aws.String(k),
			Value: aws.String(v.(string)),
		})
	}

	return result
}

func tagsToMapEMR(ts []*emr.Tag) map[string]string {
	result := make(map[string]string)
	for _, t := range ts {
		result[*t.Key] = *t.Value
	}

	return result
}

func diffTagsEMR(oldTags, newTags []*emr.Tag) ([]*emr.Tag, []*emr.Tag) {
	// First, we're creating everything we have
	create := make(map[string]interface{})
	for _, t := range newTags {
		create[*t.Key] = *t.Value
	}

	// Build the list of what to remove
	var remove []*emr.Tag
	for _, t := range oldTags {
		old, ok := create[*t.Key]
		if !ok || old != *t.Value {
			// Delete it!
			remove = append(remove, t)
		}
	}

	return expandTags(create), remove
}

func setTagsEMR(conn *emr.EMR, d *schema.ResourceData) error {
	if d.HasChange("tags") {
		oraw, nraw := d.GetChange("tags")
		o := oraw.(map[string]interface{})
		n := nraw.(map[string]interface{})
		create, remove := diffTagsEMR(expandTags(o), expandTags(n))

		// Set tags
		if len(remove) > 0 {
			log.Printf("[DEBUG] Removing tags: %s", remove)
			k := make([]*string, len(remove), len(remove))
			for i, t := range remove {
				k[i] = t.Key
			}

			_, err := conn.RemoveTags(&emr.RemoveTagsInput{
				ResourceId: aws.String(d.Id()),
				TagKeys:    k,
			})
			if err != nil {
				return err
			}
		}
		if len(create) > 0 {
			log.Printf("[DEBUG] Creating tags: %s", create)
			_, err := conn.AddTags(&emr.AddTagsInput{
				ResourceId: aws.String(d.Id()),
				Tags:       create,
			})
			if err != nil {
				return err
			}
		}
	}

	return nil
}

func expandBootstrapActions(bootstrapActions []interface{}) []*emr.BootstrapActionConfig {
	actionsOut := []*emr.BootstrapActionConfig{}

	for _, raw := range bootstrapActions {
		actionAttributes := raw.(map[string]interface{})
		actionName := actionAttributes["name"].(string)
		actionPath := actionAttributes["path"].(string)
		actionArgs := actionAttributes["args"].([]interface{})

		action := &emr.BootstrapActionConfig{
			Name: aws.String(actionName),
			ScriptBootstrapAction: &emr.ScriptBootstrapActionConfig{
				Path: aws.String(actionPath),
				Args: expandStringList(actionArgs),
			},
		}
		actionsOut = append(actionsOut, action)
	}

	return actionsOut
}

func expandInstanceGroupConfigs(instanceGroupConfigs []interface{}) []*emr.InstanceGroupConfig {
	configsOut := []*emr.InstanceGroupConfig{}

	for _, raw := range instanceGroupConfigs {
		configAttributes := raw.(map[string]interface{})
		configInstanceRole := configAttributes["instance_role"].(string)
		configInstanceCount := configAttributes["instance_count"].(int)
		configInstanceType := configAttributes["instance_type"].(string)
		configName := configAttributes["name"].(string)
		config := &emr.InstanceGroupConfig{
			Name:          aws.String(configName),
			InstanceRole:  aws.String(configInstanceRole),
			InstanceType:  aws.String(configInstanceType),
			InstanceCount: aws.Int64(int64(configInstanceCount)),
		}

		if bidPrice, ok := configAttributes["bid_price"]; ok {
			if bidPrice != "" {
				config.BidPrice = aws.String(bidPrice.(string))
				config.Market = aws.String("SPOT")
			} else {
				config.Market = aws.String("ON_DEMAND")
			}
		}

		if rawEbsConfigs, ok := configAttributes["ebs_config"]; ok {
			ebsConfig := &emr.EbsConfiguration{}

			ebsBlockDeviceConfigs := make([]*emr.EbsBlockDeviceConfig, 0)
			for _, rawEbsConfig := range rawEbsConfigs.(*schema.Set).List() {
				rawEbsConfig := rawEbsConfig.(map[string]interface{})
				ebsBlockDeviceConfig := &emr.EbsBlockDeviceConfig{
					VolumesPerInstance: aws.Int64(int64(rawEbsConfig["volumes_per_instance"].(int))),
					VolumeSpecification: &emr.VolumeSpecification{
						SizeInGB:   aws.Int64(int64(rawEbsConfig["size"].(int))),
						VolumeType: aws.String(rawEbsConfig["type"].(string)),
					},
				}
				if v, ok := rawEbsConfig["iops"].(int); ok && v != 0 {
					ebsBlockDeviceConfig.VolumeSpecification.Iops = aws.Int64(int64(v))
				}
				ebsBlockDeviceConfigs = append(ebsBlockDeviceConfigs, ebsBlockDeviceConfig)
			}
			ebsConfig.EbsBlockDeviceConfigs = ebsBlockDeviceConfigs

			config.EbsConfiguration = ebsConfig
		}

		configsOut = append(configsOut, config)
	}

	return configsOut
}

func expandConfigures(input string) []*emr.Configuration {
	configsOut := []*emr.Configuration{}
	if strings.HasPrefix(input, "http") {
		if err := readHttpJson(input, &configsOut); err != nil {
			log.Printf("[ERR] Error reading HTTP JSON: %s", err)
		}
	} else if strings.HasSuffix(input, ".json") {
		if err := readLocalJson(input, &configsOut); err != nil {
			log.Printf("[ERR] Error reading local JSON: %s", err)
		}
	} else {
		if err := readBodyJson(input, &configsOut); err != nil {
			log.Printf("[ERR] Error reading body JSON: %s", err)
		}
	}
	log.Printf("[DEBUG] Expanded EMR Configurations %s", configsOut)

	return configsOut
}

func readHttpJson(url string, target interface{}) error {
	r, err := http.Get(url)
	if err != nil {
		return err
	}
	defer r.Body.Close()

	return json.NewDecoder(r.Body).Decode(target)
}

func readLocalJson(localFile string, target interface{}) error {
	file, e := ioutil.ReadFile(localFile)
	if e != nil {
		log.Printf("[ERROR] %s", e)
		return e
	}

	return json.Unmarshal(file, target)
}

func readBodyJson(body string, target interface{}) error {
	log.Printf("[DEBUG] Raw Body %s\n", body)
	err := json.Unmarshal([]byte(body), target)
	if err != nil {
		log.Printf("[ERROR] parsing JSON %s", err)
		return err
	}
	return nil
}

func resourceAwsEMRClusterStateRefreshFunc(d *schema.ResourceData, meta interface{}) resource.StateRefreshFunc {
	return func() (interface{}, string, error) {
		conn := meta.(*AWSClient).emrconn

		log.Printf("[INFO] Reading EMR Cluster Information: %s", d.Id())
		params := &emr.DescribeClusterInput{
			ClusterId: aws.String(d.Id()),
		}

		resp, err := conn.DescribeCluster(params)

		if err != nil {
			if awsErr, ok := err.(awserr.Error); ok {
				if "ClusterNotFound" == awsErr.Code() {
					return 42, "destroyed", nil
				}
			}
			log.Printf("[WARN] Error on retrieving EMR Cluster (%s) when waiting: %s", d.Id(), err)
			return nil, "", err
		}

		emrc := resp.Cluster

		if emrc == nil {
			return 42, "destroyed", nil
		}

		if resp.Cluster.Status != nil {
			log.Printf("[DEBUG] EMR Cluster status (%s): %s", d.Id(), *resp.Cluster.Status)
		}

		status := emrc.Status
		if *status.State == "TERMINATING" || *status.State == "TERMINATED_WITH_ERRORS" {
			reason := *status.StateChangeReason
			return emrc, *status.State, fmt.Errorf("%s: %s",
				*reason.Code, *reason.Message)
		}

		return emrc, *status.State, nil
	}
}