text:
import asyncio
import logging
import time
from collections import OrderedDict
from datetime import datetime
from enum import Enum
from typing import (
    Mapping,
    Optional,
    List,
    Tuple,
    Union,
    Sequence,
    AsyncGenerator,
    Iterable,
    TYPE_CHECKING,
)

from aioredis import ConnectionClosedError, ReplyError
from aioredis.commands import streams
from aioredis.util import decode

from lightbus.client.utilities import queue_exception_checker, ErrorQueueType
from lightbus.transports.base import EventTransport, EventMessage
from lightbus.log import LBullets, L, Bold
from lightbus.serializers import ByFieldMessageSerializer, ByFieldMessageDeserializer
from lightbus.transports.redis.utilities import (
    RedisEventMessage,
    RedisTransportMixin,
    normalise_since_value,
    datetime_to_redis_steam_id,
    redis_stream_id_add_one,
    redis_stream_id_subtract_one,
    retry_on_redis_connection_failure,
)
from lightbus.utilities.async_tools import cancel
from lightbus.utilities.internal_queue import InternalQueue
from lightbus.utilities.frozendict import frozendict
from lightbus.utilities.human import human_time
from lightbus.utilities.importing import import_from_string

if TYPE_CHECKING:
    # pylint: disable=unused-import,cyclic-import
    from lightbus.config import Config
    from lightbus.client import BusClient

logger = logging.getLogger("lightbus.transports.redis")

Since = Union[str, datetime, None]

class StreamUse(Enum):
    PER_API = "per_api"
    PER_EVENT = "per_event"

    def __eq__(self, other):
        # pylint: disable=comparison-with-callable
        if isinstance(other, str):
            return self.value == other
        else:
            return super().__eq__(other)

    # Need to define this manually because we also
    # defined __eq__
    __hash__ = Enum.__hash__

class RedisEventTransport(RedisTransportMixin, EventTransport):
    """Redis Event Transport

    For a description of the protocol see https://lightbus.org/reference/protocols/event/
    """

    def __init__(
        self,
        redis_pool=None,
        *,
        service_name: str,
        consumer_name: str,
        url=None,
        serializer=ByFieldMessageSerializer(),
        deserializer=ByFieldMessageDeserializer(RedisEventMessage),
        connection_parameters: Mapping = frozendict(maxsize=100),
        batch_size=10,
        reclaim_batch_size: int = None,
        acknowledgement_timeout: float = 60,
        reclaim_interval: float = 60,
        max_stream_length: Optional[int] = 100_000,
        stream_use: StreamUse = StreamUse.PER_API,
        consumption_restart_delay: int = 5,
        consumer_ttl: int = 2_592_000,
    ):
        self.set_redis_pool(redis_pool, url, connection_parameters)
        self.batch_size = batch_size
        self.reclaim_batch_size = reclaim_batch_size if reclaim_batch_size else batch_size * 10
        self.service_name = service_name
        self.consumer_name = consumer_name
        self.acknowledgement_timeout = acknowledgement_timeout
        self.reclaim_interval = reclaim_interval
        self.max_stream_length = max_stream_length
        self.stream_use = stream_use
        self.consumption_restart_delay = consumption_restart_delay
        self.consumer_ttl = consumer_ttl
        super().__init__(serializer=serializer, deserializer=deserializer)

    @classmethod
    def from_config(
        cls,
        config: "Config",
        service_name: str = None,
        consumer_name: str = None,
        url: str = "redis://127.0.0.1:6379/0",
        connection_parameters: Mapping = frozendict(maxsize=100),
        batch_size: int = 10,
        reclaim_batch_size: int = None,
        serializer: str = "lightbus.serializers.ByFieldMessageSerializer",
        deserializer: str = "lightbus.serializers.ByFieldMessageDeserializer",
        acknowledgement_timeout: float = 60,
        reclaim_interval: Optional[float] = None,
        max_stream_length: Optional[int] = 100_000,
        stream_use: StreamUse = StreamUse.PER_API,
        consumption_restart_delay: int = 5,
        consumer_ttl: int = 2_592_000,
    ):
        serializer = import_from_string(serializer)()
        deserializer = import_from_string(deserializer)(RedisEventMessage)
        service_name = service_name or config.service_name
        consumer_name = consumer_name or config.process_name
        if isinstance(stream_use, str):
            stream_use = StreamUse[stream_use.upper()]

        if reclaim_interval is None:
            reclaim_interval = acknowledgement_timeout

        return cls(
            redis_pool=None,
            service_name=service_name,
            consumer_name=consumer_name,
            url=url,
            connection_parameters=connection_parameters,
            batch_size=batch_size,
            reclaim_batch_size=reclaim_batch_size,
            serializer=serializer,
            deserializer=deserializer,
            acknowledgement_timeout=acknowledgement_timeout,
            reclaim_interval=reclaim_interval,
            max_stream_length=max_stream_length or None,
            stream_use=stream_use,
            consumption_restart_delay=consumption_restart_delay,
            consumer_ttl=consumer_ttl,
        )

    async def send_event(self, event_message: EventMessage, options: dict):
        """Publish an event"""
        stream = self._get_stream_names(
            listen_for=[(event_message.api_name, event_message.event_name)]
        )[0]

        logger.debug(
            LBullets(
                L(
                    "Enqueuing event message {} in Redis stream {}",
                    Bold(event_message),
                    Bold(stream),
                ),
                items=dict(**event_message.get_metadata(), kwargs=event_message.get_kwargs()),
            )
        )

        # Performance: I suspect getting a connection from the connection manager each time is causing
        # performance issues. Need to confirm.
        with await self.connection_manager() as redis:
            start_time = time.time()
            native_id: bytes = await redis.xadd(
                stream=stream,
                fields=self.serializer(event_message),
                max_len=self.max_stream_length or None,
                exact_len=False,
            )

        logger.debug(
            L(
                "Enqueued event message {} in Redis in {} stream {}",
                Bold(event_message.canonical_name),
                human_time(time.time() - start_time),
                Bold(stream),
            )
        )

        event_message.native_id = native_id.decode("utf8")
        return event_message

    async def consume(
        self,
        listen_for: List[Tuple[str, str]],
        listener_name: str,
        error_queue: ErrorQueueType,
        since: Union[Since, Sequence[Since]] = "$",
        forever=True,
    ) -> AsyncGenerator[List[RedisEventMessage], None]:
        """Consume events for the given APIs"""
        self._sanity_check_listen_for(listen_for)

        consumer_group = f"{self.service_name}-{listener_name}"

        if not isinstance(since, (list, tuple)):
            # Since has been specified as a single value. Normalise it into
            # the value-per-listener format.
            since = [since] * len(listen_for)
        since = map(normalise_since_value, since)

        stream_names = self._get_stream_names(listen_for)
        # Keys are stream names, values as the latest ID consumed from that stream
        streams = OrderedDict(zip(stream_names, since))
        expected_events = {event_name for _, event_name in listen_for}

        logger.debug(
            LBullets(
                L(
                    "Consuming events as consumer {} in group {} on streams",
                    Bold(self.consumer_name),
                    Bold(consumer_group),
                ),
                items={"{} ({})".format(*v) for v in streams.items()},
            )
        )

        # Cleanup any old groups & consumers
        await self._cleanup(stream_names)

        # Here we use a queue to combine messages coming from both the
        # fetch messages loop and the reclaim messages loop.
        queue = InternalQueue(maxsize=1)
        initial_reclaiming_complete = asyncio.Event()

        async def consume_loop():
            """Regular event consuming. See _fetch_new_messages()"""
            logger.debug(
                "Will begin consuming events once the initial event reclaiming is complete"
            )
            await initial_reclaiming_complete.wait()
            logger.debug("Event reclaiming is complete, beginning to consume events")

            async for messages in self._fetch_new_messages(
                streams, consumer_group, expected_events, forever
            ):
                await queue.put(messages)
                # Wait for the queue to empty before getting trying to get another message
                await queue.join()

        retry_consume_loop = retry_on_redis_connection_failure(
            fn=consume_loop, retry_delay=self.consumption_restart_delay, action="consuming events"
        )

        async def reclaim_loop():
            """
            Reclaim messages which other consumers have failed to
            processes in reasonable time. See _reclaim_lost_messages()
            """
            while True:
                logger.debug("Checking for any events which need reclaiming")
                async for messages in self._reclaim_lost_messages(
                    stream_names, consumer_group, expected_events
                ):
                    await queue.put(messages)
                    # Wait for the queue to empty before getting trying to get another message
                    await queue.join()

                initial_reclaiming_complete.set()
                await asyncio.sleep(self.reclaim_interval)

        consume_task = None
        reclaim_task = None

        try:
            # Run the two above coroutines in their own tasks
            consume_task = asyncio.ensure_future(
                queue_exception_checker(retry_consume_loop, error_queue)
            )
            reclaim_task = asyncio.ensure_future(
                queue_exception_checker(reclaim_loop(), error_queue)
            )

            while True:
                try:
                    messages = await queue.get()
                    logger.debug(
                        f"Got batch of {len(messages)} message(s). Yielding messages to Lightbus"
                        " client"
                    )
                    yield messages
                    logger.debug(
                        f"Batch of {len(messages)} message(s) was processed by Lightbus client."
                        " Marking as done."
                    )
                    queue.task_done()
                except GeneratorExit:
                    return
        finally:
            # Make sure we cleanup the tasks we created
            await cancel(consume_task, reclaim_task)

    async def _fetch_new_messages(
        self, streams, consumer_group, expected_events, forever
    ) -> AsyncGenerator[List[EventMessage], None]:
        """Coroutine to consume new messages

        The consumption has two stages:

          1. Fetch and yield any messages this consumer is responsible for processing but has yet
             to successfully process. This can happen in cases where a message was
             previously consumed but not acknowledged (i.e. due to an error).
             This is a one-off startup stage.
          2. Wait for new messages to arrive. Yield these messages when they arrive, then
             resume waiting for messages

        See Also:

            _reclaim_lost_messages() - Another coroutine which reclaims messages which timed out
                                       while being processed by other consumers in this group

        """
        with await self.connection_manager() as redis:
            # Firstly create the consumer group if we need to
            await self._create_consumer_groups(streams, redis, consumer_group)

            # Get any messages that this consumer has yet to process.
            # This can happen in the case where the processes died before acknowledging.
            pending_messages = await redis.xread_group(
                group_name=consumer_group,
                consumer_name=self.consumer_name,
                streams=list(streams.keys()),
                # Using ID '0' indicates we want unacked pending messages
                latest_ids=["0"] * len(streams),
                timeout=None,  # Don't block, return immediately
            )

            event_messages = []
            for stream, message_id, fields in pending_messages:
                message_id = decode(message_id, "utf8")
                stream = decode(stream, "utf8")
                try:
                    event_message = self._fields_to_message(
                        fields,
                        expected_events,
                        stream=stream,
                        native_id=message_id,
                        consumer_group=consumer_group,
                    )
                except (NoopMessage, IgnoreMessage):
                    # This listener doesn't need to care about this message, so acknowledge
                    # it and move on with our lives
                    await redis.xack(stream, consumer_group, message_id)
                    continue

                logger.debug(
                    LBullets(
                        L(
                            "⬅ Receiving pending event {} on stream {}",
                            Bold(message_id),
                            Bold(stream),
                        ),
                        items=dict(
                            **event_message.get_metadata(), kwargs=event_message.get_kwargs()
                        ),
                    )
                )
                event_messages.append(event_message)

            if event_messages:
                yield event_messages

            # We've now cleaned up any old messages that were hanging around.
            # Now we get on to the main loop which blocks and waits for new messages

            while True:
                # Fetch some messages.
                # This will block until there are some messages available
                try:
                    stream_messages = await redis.xread_group(
                        group_name=consumer_group,
                        consumer_name=self.consumer_name,
                        streams=list(streams.keys()),
                        # Using ID '>' indicates we only want new messages which have not
                        # been passed to other consumers in this group
                        latest_ids=[">"] * len(streams),
                        count=self.batch_size,
                    )
                except asyncio.CancelledError:
                    # We need to manually close the connection here otherwise the aioredis
                    # pool will emit warnings saying that this connection still has pending
                    # commands (i.e. the above blocking pop)
                    redis.close()
                    raise

                # Handle the messages we have received
                event_messages = []
                for stream, message_id, fields in stream_messages:
                    message_id = decode(message_id, "utf8")
                    stream = decode(stream, "utf8")
                    try:
                        event_message = self._fields_to_message(
                            fields,
                            expected_events,
                            stream=stream,
                            native_id=message_id,
                            consumer_group=consumer_group,
                        )
                    except (NoopMessage, IgnoreMessage):
                        # This listener doesn't need to care about this message, so acknowledge
                        # it and move on with our lives
                        logger.debug(
                            f"Ignoring NOOP event with ID {message_id} discovered during streaming"
                            " of messages"
                        )
                        await redis.xack(stream, consumer_group, message_id)
                        continue

                    logger.debug(
                        LBullets(
                            L(
                                "⬅ Received new event {} on stream {}",
                                Bold(message_id),
                                Bold(stream),
                            ),
                            items=dict(
                                **event_message.get_metadata(), kwargs=event_message.get_kwargs()
                            ),
                        )
                    )

                    event_messages.append(event_message)

                if event_messages:
                    yield event_messages

                if not forever:
                    return

    async def _reclaim_lost_messages(
        self, stream_names: List[str], consumer_group: str, expected_events: set
    ) -> AsyncGenerator[List[EventMessage], None]:
        """Reclaim batches of messages that other consumers in the group failed to acknowledge within a timeout.

        The timeout period is specified by the `acknowledgement_timeout` option.
        """
        with await self.connection_manager() as redis:
            for stream in stream_names:

                old_messages = True
                reclaim_from = None

                # Keep pulling reclaimable messages from Redis until there are none left
                while old_messages:
                    # reclaim_from keeps track of where we are up to in our fetching
                    # of messages
                    if not reclaim_from:
                        # This is our first iteration, so fetch from the start of time
                        reclaim_from = "-"
                    else:
                        # This is a subsequent iteration. XPENDING's 'start' parameter is inclusive,
                        # so we need to add one to the reclaim_from value to ensure we don't get a message
                        # we've already seen
                        reclaim_from = redis_stream_id_add_one(reclaim_from)

                    # Fetch the next batch of messages
                    try:
                        old_messages = await redis.xpending(
                            stream, consumer_group, reclaim_from, "+", count=self.reclaim_batch_size
                        )
                    except ReplyError as e:
                        if "NOGROUP" in str(e):
                            # Group or consumer doesn't exist yet, so stop processing for this loop.
                            break
                        else:
                            raise

                    timeout = self.acknowledgement_timeout * 1000
                    event_messages = []

                    # Try to claim each messages
                    for (
                        message_id,
                        consumer_name,
                        ms_since_last_delivery,
                        num_deliveries,
                    ) in old_messages:
                        message_id = decode(message_id, "utf8")
                        consumer_name = decode(consumer_name, "utf8")
                        reclaim_from = message_id

                        # This 'if' is not strictly required as the subsequent call to xclaim
                        # will honor the timeout parameter. However, using this if here allows
                        # for more sane logging from the point of view of the user. Without it
                        # we would report that we were trying to claim messages which were
                        # clearly not timed out yet.
                        if ms_since_last_delivery > timeout:
                            logger.info(
                                L(
                                    "Found timed out event {} in stream {}. Abandoned by {}."
                                    " Attempting to reclaim...",
                                    Bold(message_id),
                                    Bold(stream),
                                    Bold(consumer_name),
                                )
                            )

                            # *Try* to claim the message...
                            result = await redis.xclaim(
                                stream, consumer_group, self.consumer_name, int(timeout), message_id
                            )

                            # Parse each message we managed to claim
                            for _, fields in result:
                                # Note that sometimes we will claim a message and it will be 'nil'.
                                # In this case the result will be (None, {}). We therefore do not
                                # rely on the values we get back from XCLAIM.
                                # I suspect this may happen if a stream has been trimmed, thereby causing
                                # un-acknowledged messages to be deleted from Redis.
                                try:
                                    event_message = self._fields_to_message(
                                        fields,
                                        expected_events,
                                        stream=stream,
                                        native_id=message_id,
                                        consumer_group=consumer_group,
                                    )
                                except (NoopMessage, IgnoreMessage):
                                    # This listener doesn't need to care about this message, so acknowledge
                                    # it and move on with our lives
                                    logger.debug(
                                        f"Ignoring NOOP event with ID {message_id} discovered"
                                        " during event reclaiming"
                                    )
                                    await redis.xack(stream, consumer_group, message_id)
                                    continue

                                logger.debug(
                                    LBullets(
                                        L(
                                            "⬅ Reclaimed timed out event {} on stream {}. Abandoned"
                                            " by {}.",
                                            Bold(message_id),
                                            Bold(stream),
                                            Bold(consumer_name),
                                        ),
                                        items=dict(
                                            **event_message.get_metadata(),
                                            kwargs=event_message.get_kwargs(),
                                        ),
                                    )
                                )
                                event_messages.append(event_message)

                    # And yield our batch of messages
                    if event_messages:
                        yield event_messages

    async def acknowledge(self, *event_messages: RedisEventMessage):
        """Acknowledge that a message has been successfully processed"""
        with await self.connection_manager() as redis:
            p = redis.pipeline()
            for event_message in event_messages:
                p.xack(event_message.stream, event_message.consumer_group, event_message.native_id)
                logging.debug(
                    f"Preparing to acknowledge message {event_message.id} (Native ID:"
                    f" {event_message.native_id})"
                )

            logger.debug(
                f"Batch acknowledging successful processing of {len(event_messages)} message."
            )
            await p.execute()

    async def history(
        self,
        api_name,
        event_name,
        start: datetime = None,
        stop: datetime = None,
        start_inclusive: bool = True,
        batch_size: int = 100,
    ) -> AsyncGenerator[EventMessage, None]:
        """Retrieve historical events for the given API

        Will not have any impact on existing consumer groups.
        """
        redis_start = datetime_to_redis_steam_id(start) if start else "-"
        redis_stop = datetime_to_redis_steam_id(stop) if stop else "+"

        if start and not start_inclusive:
            redis_start = redis_stream_id_add_one(redis_start)

        stream_name = self._get_stream_names([(api_name, event_name)])[0]

        logger.debug(
            f"Getting history for stream {stream_name} from {redis_start} ({start}) "
            f"to {redis_stop} ({stop}) in batches of {batch_size}"
        )

        with await self.connection_manager() as redis:
            messages = True
            while messages:
                messages = await redis.xrevrange(
                    stream_name, redis_stop, redis_start, count=batch_size
                )
                if not messages:
                    return
                for message_id, fields in messages:
                    message_id = decode(message_id, "utf8")
                    redis_stop = redis_stream_id_subtract_one(message_id)
                    try:
                        event_message = self._fields_to_message(
                            fields,
                            expected_event_names={event_name},
                            stream=stream_name,
                            native_id=message_id,
                            consumer_group=None,
                        )
                    except (NoopMessage, IgnoreMessage):
                        logger.debug(
                            f"Ignoring NOOP event with ID {message_id} discovered during fetching"
                            " of event history"
                        )
                    else:
                        yield event_message

    async def lag(
        self,
        api_name: str,
        event_name: str,
        listener_name: str,
        include_pending: bool = True,
        max_count: Optional[int] = 100,
    ) -> int:
        """How far behind is the given listener for the given api/event?

        This must fetch all lagging messages from Redis, which means
        this cannot be particularly performant for determining the size of large
        lags. The ideal is therefore to set max_count to a sane minimum, thereby
        minimising the number of messages which need to be pulled back.
        Note however, that this method will never return a value larger than
        max_count + the total pending messages.

        If max_count is None then the full lag will be calculated.
        """
        consumer_group = f"{self.service_name}-{listener_name}".encode("utf8")
        stream_name = self._get_stream_names([(api_name, event_name)])[0]

        with await self.connection_manager() as redis:
            try:
                # Get the groups for the given stream
                groups = await redis.xinfo_groups(stream_name)
            except ReplyError as e:
                if "no such key" in str(e):
                    # Steam doesn't exist, so group doesn't exist
                    groups = []
                else:
                    raise

            for group in groups:
                # Find the group we care about
                if group[b"name"] == consumer_group:
                    if max_count is not None:
                        # We should normally receive the last delivered message in addition to any lagging
                        # messages We take that into account here and is tested by test_lag_max_count_set()
                        max_count += 1

                    # This is the part with poor performance. Get all the lagging messages.
                    messages = await redis.xrange(
                        stream=stream_name, start=group[b"last-delivered-id"], count=max_count
                    )

                    # Messages found, exit our for loop
                    break
            else:
                # For loop didn't find any matching consumer groups, return 0
                return 0

        first_message_id = messages[0][0] if messages else None
        total_messages = len(messages)

        if first_message_id == group[b"last-delivered-id"]:
            # The last delivered message is among the results returned, so
            # take that off our count
            lag = total_messages - 1
        elif max_count is not None and total_messages == max_count:
            # The last delivered message has been deleted, and we received the max_count number of results,
            # which is actually one more than we expected because we incremented max_count above.
            # Therefore fix this by subtracting one
            lag = total_messages - 1
        else:
            # The last delivered message has been deleted,
            # so all returned messages are lagging. We retrieved all messages
            # (rather than limiting with max_count) therefore no need to
            # subtract anything
            lag = total_messages

        if include_pending:
            lag += group[b"pending"]

        return lag

    async def _create_consumer_groups(self, streams, redis, consumer_group):
        """Ensure the consumer groups exist

        This is means we have to ensure the streams exist too
        """
        for stream, since in streams.items():
            if not await redis.exists(stream):
                # Add a noop to ensure the stream exists
                # TODO: We can now use MKSTREAM, change this logic
                #       Documented here: https://redis.io/topics/streams-intro
                await redis.xadd(stream, fields={"": ""})

            try:
                # Create the group (it may already exist)
                await redis.xgroup_create(stream, consumer_group, latest_id=since)
            except ReplyError as e:
                if "BUSYGROUP" in str(e):
                    # Already exists
                    pass
                else:
                    raise

    async def _cleanup(self, stream_names: List[str]):
        """Cleanup old consumers and groups

        A group will be deleted if it contains no consumers.

        A consumer will be deleted if it has been idle for more than consumer_ttl.
        """
        if not self.consumer_ttl:
            # Don't do the cleanup if no TTL is given, consider this to mean
            # cleanup is disabled
            return

        with await self.connection_manager() as redis:
            # For every stream key...
            for stream_name in stream_names:
                consumers: List[Tuple[str, str]] = []

                # Get all the groups for that key...
                try:
                    groups = await redis.xinfo_groups(stream_name)
                except ReplyError as e:
                    if "ERR no such key" in str(e):
                        # Steam doesn't exist yet
                        groups = []
                    else:
                        raise

                for group in groups:
                    active_consumers = 0
                    group_name = group[b"name"]

                    try:
                        consumers = await redis.xinfo_consumers(stream_name, group_name)
                    except ReplyError as e:
                        # Group has already vanished,
                        # perhaps something else is doing a cleanup
                        if "NOGROUP" in str(e):
                            continue

                    # Get all the consumers for that group
                    for consumer in consumers:
                        consumer_name = consumer[b"name"]
                        idle_seconds = consumer[b"idle"] / 1000

                        # And delete the consumer if they have not re-started
                        # listening for self.consumer_ttl seconds
                        if idle_seconds >= self.consumer_ttl:
                            logger.debug(
                                f"Cleaning up consumer {consumer_name} in group {group_name} on"
                                f" stream {stream_name}. The consumer has been idle for"
                                f" {idle_seconds} seconds, which is more than the consumer TTL of"
                                f" {self.consumer_ttl}"
                            )
                            await redis.xgroup_delconsumer(stream_name, group_name, consumer_name)
                        else:
                            active_consumers += 1

                    # If no active consumers were found for this group, then delete the entire group
                    # on the grounds that it is no longer used and can be cleaned up.
                    if not active_consumers:
                        # We do this atomically using a lua script. This avoids race conditions
                        # whereby a new consumer comes into existence the moment before we delete the group
                        try:
                            await redis.eval(
                                ATOMIC_DESTROY_CONSUMER_GROUP, [stream_name], [group_name]
                            )
                        except ReplyError as e:
                            if "NOGROUP" in str(e):
                                # Already deleted
                                pass

    def _fields_to_message(
        self,
        fields: dict,
        expected_event_names: Iterable[str],
        stream: str,
        native_id: str,
        consumer_group: Optional[str],
    ) -> Optional[RedisEventMessage]:
        """Convert a dict of Redis message fields into a RedisEventMessage"""

        if not fields:
            # Redis returned us a message of 'nil'
            raise NoopMessage()

        if tuple(fields.items()) == ((b"", b""),):
            # Is a noop message, ignore
            raise NoopMessage()

        message = self.deserializer(
            fields, stream=stream, native_id=native_id, consumer_group=consumer_group
        )

        want_message = ("*" in expected_event_names) or (message.event_name in expected_event_names)
        if self.stream_use == StreamUse.PER_API and not want_message:
            # Only care about events we are listening for. If we have one stream
            # per API then we're probably going to receive some events we don't care about.
            logger.debug(
                f"Ignoring message for unneeded event: {message}. "
                f"Only listening for {', '.join(expected_event_names)}"
            )
            raise IgnoreMessage()
        return message

    def _get_stream_names(self, listen_for):
        """Convert a list of api names & event names into stream names

        The format of these names will vary based on the stream_use setting.
        """
        stream_names = []
        for api_name, event_name in listen_for:
            if self.stream_use == StreamUse.PER_EVENT:
                stream_name = f"{api_name}.{event_name}:stream"
            elif self.stream_use == StreamUse.PER_API:
                stream_name = f"{api_name}.*:stream"
            else:
                raise ValueError(
                    "Invalid value for stream_use config option. This should have been caught "
                    "during config validation."
                )
            if stream_name not in stream_names:
                stream_names.append(stream_name)
        return stream_names

class NoopMessage(Exception):
    pass

class IgnoreMessage(Exception):
    pass

# See RedisEventTransport._cleanup()
ATOMIC_DESTROY_CONSUMER_GROUP = """
local stream_name = KEYS[1]
local group_name = ARGV[1]
local consumers = redis.call('xinfo', 'consumers', stream_name, group_name)
if table.getn(consumers) == 0 then
    redis.call('xgroup', 'destroy', stream_name, group_name)
end
"""

def parse_messages(messages):
    # TODO: Remove and bump required aioredis version when merged:
    #       https://github.com/aio-libs/aioredis/pull/723
    if messages is None:
        return []

    parsed_messages = []
    for message in messages:
        if message is None:
            # In some conditions redis will return a NIL message
            parsed_messages.append((None, {}))
        else:
            mid, values = message
            parsed_messages.append((mid, streams.fields_to_dict(values)))

    return parsed_messages

streams.parse_messages = parse_messages


text:
require(['cdf/AddIn',
        'cdf/Dashboard',
        'cdf/dashboard/Sprintf',
        'amd!cdf/lib/underscore',
        'cdf/lib/jquery',
        'cdf/lib/CCC/protovis-compat!',
        'amd!cdf/lib/datatables',
        'css!./dataBar'],

function(AddIn, Dashboard, sprintf, _, $, pv) {
    var stackedDataBar = {
        name: "stackedDataBar",
        label: "Stacked Data Bar",
        defaults: {
            width: undefined,
            widthRatio: 1,
            height: undefined,
            align: null,
            startColor: "#55A4D6",
            endColor: "#448FC8",
            backgroundImage: undefined,
            stroke: null,
            max: undefined,
            min: undefined,
            includeValue: false,
            absValue: true,
            stacked: false,
            stackedColors: ["#AAA","#BBB","#CCC"],
            valueFormat: function(v, format, st, opt) {
                return "" + sprintf(format || "%.0f", v);
            }
        },
        init: function() {
            $.fn.dataTableExt.oSort[this.name + '-asc'] = $.fn.dataTableExt.oSort['numeric-asc'];
            $.fn.dataTableExt.oSort[this.name + '-desc'] = $.fn.dataTableExt.oSort['numeric-desc'];
        },
        implementation: function(tgt, st, opt) {

            if ( opt.stacked ){

                //validate if we have values to render
                if ( st.value.length === 0 ) {
                    $(tgt).empty();
                    return
                };

                var colors = opt.stackedColors,
                    lColors = colors.length,
                    values = st.value.split(','),
                    length = values.length,
                    total = 0;

                //the colors available will be repeated to all values
                if ( lColors < length ){
                    for ( var i = 0; i <= length; i++ ){
                        var j = i;
                        if ( j > lColors ){
                            j = j - lColors - 1;
                            colors.push(colors[j]);
                        }
                    }
                }

                //determine width of bars
                _.each(values,function(value){
                    total = total + parseFloat(value);
                });

                var ratio = [];
                _.each(values,function(value){
                    var perc = value / total;
                    ratio.push(perc);
                });

                //determine cell size
                var cellSettings = this.cellSettings(tgt,opt);
                var ph = cellSettings[0];
                var wtmp = cellSettings[1] - 1;
                var htmp = cellSettings[2];
                var hasSVG = cellSettings[3];

                ph.css('min-width', ph.width() + 'px');

                //build stacked bars
                for ( var i = 0; i < length; i++ ){
                    var wbar = Math.floor(wtmp * ratio[i]);
                    if ( wbar > 0 ){
                        var phBar = $("

").addClass('stackedBar').appendTo(ph);
                        phBar.width(wbar);
                        phBar.css({'background':colors[i], "display": "block", "float": "left"});
                        //if we want to show values for each bar
                        if(opt.includeValue) {
                            var valueStr = opt.valueFormat(values[i], st.colFormat, st, opt);
                            if ( valueStr !== "0" ) {
                                var valph = $("").addClass('value');
                                valph.append(valueStr);
                                phBar.append(valph);
                                phBar.find('span').width(wbar - 4);
                            }
                        }
                    }
                }
            } else {
                var tblMax = Math.max.apply(Math,st.tableData.map(function(e) {
                    return e[st.colIdx];
                })),
                tblMin = Math.min.apply(Math,st.tableData.map(function(e) {
                    return e[st.colIdx];
                }));

                var optMax = parseFloat(opt.max);
                var optMin = parseFloat(opt.min);

                var isValidNumber = function(nr) {
                    return _.isNumber(nr) && isFinite(nr);
                };

                var validMaxValue = isValidNumber(optMax);
                var validMinValue = isValidNumber(optMin);

                if(opt.absValue) {
                    var max = (validMaxValue == true) ? optMax : Math.max(Math.abs(tblMax), Math.abs(tblMin)),
                        min = (validMinValue == true) ? optMin : 0,
                        val = Math.abs(parseFloat(st.value));
                        min = Math.max(min,0);
                } else {
                    var max = (validMaxValue == true) ? optMax : Math.max(0, tblMax),
                        min = (validMinValue == true) ? optMin : Math.min(0, tblMin),
                        val = parseFloat(st.value);
                }

                //determine cell size
                var cellSettings = this.cellSettings(tgt,opt);
                var ph = cellSettings[0];
                var wtmp = cellSettings[1];
                var htmp = cellSettings[2];
                var hasSVG = cellSettings[3];

                var leftVal = Math.min(val, 0),
                    rightVal = Math.max(val, 0),
                    options = {
                        scale: 100,
                        wtmp: wtmp,
                        htmp: htmp,
                        align: opt.align,
                        barHeight: 100,
                        r: rightVal,
                        l: leftVal,
                        hasSVG: hasSVG,
                        target: ph.get(0),
                        processVal: function(val) { return val + '%'; }
                    },
                    c;
                // if we have SVG, and wtmp is a string, so probably a percentage width, will use percentage calculations
                if(hasSVG && typeof wtmp === 'string') {
                    c = this.drawPaper(min, max, options);
                } else {
                    // falling back to the default
                    options.processVal = function(val) { return val; };
                    options.scale = wtmp;
                    options.barHeight = htmp;
                    options.legacy = true;
                    c = this.drawPaper(min, max, options);
                }

                c.attr({
                    fill: opt.backgroundImage ? "url('" + opt.backgroundImage + "')" : "90-" + opt.startColor + "-" + opt.endColor,
                    stroke: opt.stroke,
                    title: "Value: " + st.value
                });

                if(opt.includeValue) {
                    var valueStr = opt.valueFormat(st.value, st.colFormat, st, opt);
                    var valph = $("").addClass('value');
                    valph.append(valueStr);
                    if(options.legacy) {
                      valph.appendTo(ph);
                    } else if(hasSVG && opt.align == "right") {
                        valph.addClass('alignRight').appendTo(ph);
                        ph.find("svg").css('float', 'right');
                    } else {
                        valph.prependTo(ph);
                    }
                }
            }
        },
        cellSettings: function(tgt,opt){
            var hasSVG = !!(
                document.createElementNS &&
                document.createElementNS('http://www.w3.org/2000/svg', 'svg').createSVGRect
            );

            var cell = $(tgt).empty();
            var ph = $("

").addClass('dataBarContainer').appendTo(cell);
            var wtmp = opt.width || ph.width();
            if(typeof wtmp === 'string') {
                if(!hasSVG) {
                    wtmp = ph.width() * opt.widthRatio;
                } else {
                    var parsedWidth = parseFloat(wtmp);
                    // if we have a widthRatio != 1, we want to apply it to a 100% width
                    parsedWidth = ((opt.widthRatio != 1 && parsedWidth >= 98) ? 100 : parsedWidth) * opt.widthRatio;
                    wtmp = parsedWidth + "%";
                }
            } else {
                wtmp *= opt.widthRatio;
            }
            var htmp = opt.height || ph.height();
            return [ph,wtmp,htmp,hasSVG];
        },
        drawPaper: function(min, max, opts) {
            // xx = x axis
            var xx = pv.Scale.linear(min,max).range(0, opts.scale);

            var paper = Raphael(
                opts.target,
                opts.legacy ? xx(Math.min(opts.r,max)) - xx(min) : opts.wtmp,
                opts.htmp);
            if(!opts.legacy && opts.hasSVG && opts.align == "right") {
                return paper.rect(
                opts.processVal(xx(max) - xx(opts.r)),
                opts.processVal(0),
                opts.processVal(xx(opts.r) - xx(opts.l)),
                opts.processVal(opts.barHeight));
            }
            return paper.rect(
                opts.processVal(xx(opts.l)),
                opts.processVal(0),
                opts.processVal(xx(opts.r) - xx(opts.l)),
                opts.processVal(opts.barHeight));
        }
    };

    Dashboard.registerGlobalAddIn("Table", "colType",  new AddIn(stackedDataBar));

    return stackedDataBar;
});

redpajama_set_name: RedPajamaGithub


text:
<!doctype html>

	<meta charset="utf-8">
	<meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1">
	<title>raa</title>
	<meta name="description" content="">
	<meta name="viewport" content="width=device-width, initial-scale=1">

	<link rel="stylesheet" href="../css/bulma.min.css">
	<link rel="stylesheet" href="../css/main.css">
	<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/4.6.3/css/font-awesome.min.css">

EDUC 766

Instructional Strategies & Assessment Methods

EDUC 765

EDUC 766

EDUC 767

EDUC 768

						Overview
						Alignment
						Assessment

						Activities
						Reflection

Alignment

An alignment chart is used to diagram instructional activities and assessment strategies. It also shows connections between objectives (what learners will be able to do), types of activities (what learners will be doing), and assessment (how the learners will be measured). Beyond the clear connections between the objectives, activities, and objectives, the alignment chart provides a map of the instruction plan and allows the designer to show how the instruction will happen, how learners will be assessed, and how the instructional objectives are met.

Problem Identification

The project aims to meet students' information and research needs in a manner that allows for continual assistance whether or not a librarian is available.

Delivery Options

This project aims to create an organized, concise, and informative online platform where college students can independently answer simple research and library-related questions through the use of text, multimedia, and interactive content.

							TERMINAL OBJECTIVE: Identify library services and library collections.

							Enabling Objectives
							Assessment Idea
							Absorb Activity
							Do Activity
							Connect Activity

							Identify library services and their functions
							Self-assessment (Matching and multiple choice questions questions)
							Reading. Slide show. Video.
							Practice Activity (Matching and multiple choice questions questions)
							n/a

							Locate library collections
							Use a library floor map to answer self-assessment questions (Multiple choice and matching)
							Reading. Slide show. Video.
							Practice Activity (Multiple choice and matching)
							Job aide (Floor map)

							Explain how library collections are organized
							Self-assessment (Matching and multiple choice questions)
							Reading. Slide show. Video.
							Practice Activity (Matching and multiple choice questions)
							Job aide (Library of Congress Classification guide)

							Describe the use of the search tools in the library website
							Self-assessment (Matching, fill-in-blank, and multiple choice questions)
							Reading. Slide show. Video.
							Practice Activity (Matching, fill-in-blank, and multiple choice questions)
							Ponder Activity (Rhetorical question)

							TERMINAL OBJECTIVE: Find sources and information using the library catalog and library resources.

							Enabling Objectives
							Assessment Idea
							Absorb Activity
							Do Activity
							Connect Activity

							Create a list of keywords from a given topic
							Self-assessment (Fill-in-blank and multiple choice questions). Performance task (Determine keywords that will return similar search results to those found in provided image)
							Reading. Slide show. Video.
							Practice Activity (Fill-in-blank and multiple choice questions)
							Ponder Activity (Rhetorical question; Cite-example activity). Performance task (Determine keywords that will return similar search results to those found in provided image)

							Experiment with various search queries using keywords
							Self-assessment (True/False, multiple choice, and fill-in-blank questions)
							Reading. Slide show. Video.
							Practice Activity (True/False, multiple choice, and fill-in-blank questions)
							Ponder Activity (Rhetorical question; Cite-example activity)

							Find sources using one or more descriptors (e.g. title, author, journal title)
							Self-assessment (Multiple choice, true/false, and matching questions). Performance Task (Given a citation, find an item in the catalog)
							Reading. Slide show. Video.
							Practice Activity (Multiple choice, true/false, and matching questions). Performance Task (Given a citation, find an item in the catalog)
							Ponder Activity (Rhetorical question; Cite-example activity)

							Match search strategies and information needs with appropriate search tools
							Self-assessment (Matching, multiple choice, and true/false questions). Performance task (Compare and contrast search results in Google Scholar and library catalog)
							Reading. Slide show. Video.
							Practice Activity (Matching, multiple choice, and true/false questions)
							Ponder Activity (Rhetorical Question; Cite-example activity). Performance task (Compare and contrast search results in Google Scholar and library catalog)

							Use the Library of Congress Classification System to locate sources
							Self-assessment (Sequence, multiple choice, and true/false questions)
							Reading. Slide show. Video.
							Practice Activity (Sequence, multiple choice, and true/false questions)
							Ponder Activity (Rhetorical question; Cite-example activity)

							TERMINAL OBJECTIVE: Develop a research question/topic.

							Enabling Objectives
							Assessment Idea
							Absorb Activity
							Do Activity
							Connect Activity

							Identify a research question/topic
							Performance task (Identify the thesis/argument of a paper by reading the abstract). Self-assessment (Multiple choice and true/false questions)
							Reading. Slide show. Video.
							Practice Activity (Multiple choice and true/false questions). Performance task (Identify the thesis/argument of a paper by reading the abstract).
							Ponder Activity (Rhetorical question; Cite-example activity)

							Break down complex questions/topics into simple ones in order to limit scope
							Self-assessment (Fill-in-blank, matching, and multiple choice questions). Performance task (Using Boolean operators, narrow or broaden search. Try to match the provided target number of results)
							Reading. Slide show. Video.
							Practice Activity (Fill-in-blank, matching, and multiple choice questions). Performance task (Using Boolean operators, narrow or broaden search. Try to match the provided target number of results)
							Ponder Activity (Rhetorical question; Cite-example activity)

							Locate sources (scholarly or popular) for background information on a topic
							Self-assessment (Matching, multiple choice, and true/false questions)
							Reading. Slide show. Video.
							Practice Activity (Matching, multiple choice, and true/false questions)
							Ponder Activity (Rhetorical question)

							Use sources (scholarly or popular) to refine a topic
							Self-assessment (Multiple choice and matching questions)
							Reading. Slide show. Video.
							Practice Activity (Multiple choice and matching questions)
							Ponder Activity (Rhetorical question)

							Use sources (scholarly or popular) to support an argument
							Self-assessment (Multiple choice and true/false questions)
							Reading. Slide show. Video.
							Practice Activity (Multiple choice and true/false questions)
							Ponder Activity (Rhetorical question)

							Develop research questions based on information gaps or information already gathered
							Self-assessment (Multiple choice and true/false questions)
							Reading. Slide show. Video.
							Practice Activity (Multiple choice and true/false questions)
							Ponder Activity (Rhetorical question)

							TERMINAL OBJECTIVE: Use qualitative methods to identify authoritative sources of information.

							Enabling Objectives
							Assessment Idea
							Absorb Activity
							Do Activity
							Connect Activity

							Define different aspects of authority (subject expertise, reputation, professional knowledge, etc.)
							Self-assessment (Matching and multiple choice questions)
							Reading. Slide show. Video.
							Practice Activity (Matching and multiple choice questions)
							Ponder Activity (Rhetorical question)

							Explain how researchers influence and shape scholarly conversations
							Self-assessment (Multiple choice and true/false questions)
							Reading. Slide show. Video.
							Practice Activity (Multiple choice and true/false questions)
							Ponder Activity (Rhetorical question)

							Explain how a work does not represent the whole perspective on an issue
							Self-assessment (Multiple choice and true/false questions)
							Reading. Slide show. Video.
							Practice Activity (Multiple choice and true/false questions)
							Ponder Activity (Rhetorical question)

							Identify authoritative sources and content in formal and informal venues
							Self-assessment (Multiple choice, true/false, and matching questions)
							Reading. Slide show. Video.
							Practice Activity (Multiple choice, true/false, and matching questions)
							Ponder Activity (Rhetorical question)

							Interpret quantitative measures of authority
							Self-assessment (Multiple choice, matching, and true/false questions)
							Reading. Slide show. Video.
							Practice Activity (Multiple choice, matching, and true/false questions)
							Ponder Activity (Rhetorical question)

							TERMINAL OBJECTIVE: Explain the importance of attribution and copyright laws.

							Enabling Objectives
							Assessment Idea
							Absorb Activity
							Do Activity
							Connect Activity

							Articulate and distinguish the meanings of copyright, fair use, open access and the public domain
							Self-assessment (Matching and multiple choice questions)
							Reading. Slide show. Video.
							Practice Activity (Matching and multiple choice questions)
							Ponder Activity (Rhetorical question)

							Summarize the importance of intellectual property issues
							Self-assessment (Matching, true/false, multiple choice questions)
							Reading. Slide show. Video.
							Practice Activity (Matching, true/false, multiple choice questions)
							Ponder Activity (Rhetorical question)

							Give credit to original ideas through attribution
							Self-assessment (Multiple choice and true/false questions)
							Reading. Slide show. Video.
							Practice Activity (Multiple choice and true/false questions)
							Ponder Activity (Rhetorical question)

							Use a given citation style
							Self-assessment (Sequence, multiple choice, and true/false questions)
							Reading. Slide show. Video.
							Practice Activity (Sequence, multiple choice, and true/false questions)
							Ponder Activity (Rhetorical question; Cite-example activity)

						Overview

						Alignment

						Assessment

						Activities

						Reflection


text:
-- phpMyAdmin SQL Dump
-- version 4.2.11
-- http://www.phpmyadmin.net
--
-- Host: 127.0.0.1
-- Generation Time: Mar 09, 2015 at 07:25 AM
-- Server version: 5.6.21
-- PHP Version: 5.6.3

SET SQL_MODE = "NO_AUTO_VALUE_ON_ZERO";
SET time_zone = "+00:00";

/*!40101 SET @OLD_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT */;
/*!40101 SET @OLD_CHARACTER_SET_RESULTS=@@CHARACTER_SET_RESULTS */;
/*!40101 SET @OLD_COLLATION_CONNECTION=@@COLLATION_CONNECTION */;
/*!40101 SET NAMES utf8 */;

--
-- Database: `a1`
--

-- --------------------------------------------------------

--
-- Table structure for table `clients`
--

CREATE TABLE IF NOT EXISTS `clients` (
`id` int(11) NOT NULL,
  `company` varchar(255) NOT NULL,
  `email` varchar(255) NOT NULL,
  `name` varchar(255) NOT NULL,
  `rate` decimal(10,0) NOT NULL DEFAULT '25',
  `image` mediumblob,
  `description` text,
  `img_name` text
) ENGINE=InnoDB AUTO_INCREMENT=4 DEFAULT CHARSET=latin1;

--
-- Dumping data for table `clients`
--

INSERT INTO `clients` (`id`, `company`, `email`, `name`, `rate`, `image`, `description`, `img_name`) VALUES
(1, 'Jim Parry inc.', 'jparry@bcit.ca', 'Jim Parry', '25', NULL, 'Comp 4711 instructor', NULL),
(2, 'Electronic Arts', 'joe@ea.com', 'Joe Blow', '25', NULL, NULL, NULL),
(3, '{company}', '{email}', '{name}', '0', NULL, NULL, NULL);

-- --------------------------------------------------------

--
-- Table structure for table `jobs`
--

CREATE TABLE IF NOT EXISTS `jobs` (
  `id` int(11) NOT NULL,
  `note` text,
  `status` enum('outstanding','invoice','paid') NOT NULL DEFAULT 'outstanding',
  `hours` decimal(10,2) NOT NULL,
  `clientId` int(11) NOT NULL
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

--
-- Dumping data for table `jobs`
--

INSERT INTO `jobs` (`id`, `note`, `status`, `hours`, `clientId`) VALUES
(1, 'HELLO WORLD', 'outstanding', '0.25', 1);

-- --------------------------------------------------------

--
-- Table structure for table `users`
--

CREATE TABLE IF NOT EXISTS `users` (
  `username` varchar(255) NOT NULL,
  `password` varchar(255) NOT NULL,
  `type` enum('admin','client','user') NOT NULL DEFAULT 'user',
  `clientId` int(11) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

--
-- Dumping data for table `users`
--

INSERT INTO `users` (`username`, `password`, `type`, `clientId`) VALUES
('admin', 'P@$$w0rd', 'admin', NULL),
('jlparry', 'P@$$w0rd', 'client', 1);

--
-- Indexes for dumped tables
--

--
-- Indexes for table `clients`
--
ALTER TABLE `clients`
 ADD PRIMARY KEY (`id`), ADD UNIQUE KEY `company` (`company`);

--
-- Indexes for table `jobs`
--
ALTER TABLE `jobs`
 ADD PRIMARY KEY (`id`);

--
-- Indexes for table `users`
--
ALTER TABLE `users`
 ADD PRIMARY KEY (`username`);

--
-- AUTO_INCREMENT for dumped tables
--

--
-- AUTO_INCREMENT for table `clients`
--
ALTER TABLE `clients`
MODIFY `id` int(11) NOT NULL AUTO_INCREMENT,AUTO_INCREMENT=4;
/*!40101 SET CHARACTER_SET_CLIENT=@OLD_CHARACTER_SET_CLIENT */;
/*!40101 SET CHARACTER_SET_RESULTS=@OLD_CHARACTER_SET_RESULTS */;
/*!40101 SET COLLATION_CONNECTION=@OLD_COLLATION_CONNECTION */;


text:
namespace Merchello.Core.Services
{
    using System;
    using System.Collections.Generic;
    using Models;
    using Umbraco.Core.Services;

    /// <summary>
    /// Defines the AppliedPaymentService
    /// </summary>
    internal interface IAppliedPaymentService : IService
    {
        /// <summary>
        /// Creates and saves an AppliedPayment
        /// </summary>
        /// <param name="paymentKey">The PaymentGatewayProvider key</param>
        /// <param name="invoiceKey">The invoice 'key'</param>
        /// <param name="appliedPaymentType">The applied payment type</param>
        /// <param name="description">The description of the payment application</param>
        /// <param name="amount">The amount of the payment to be applied</param>
        /// <param name="raiseEvents">Optional boolean indicating whether or not to raise events</param>
        /// <returns>An <see cref="IAppliedPayment"/></returns>
        IAppliedPayment CreateAppliedPaymentWithKey(Guid paymentKey, Guid invoiceKey, AppliedPaymentType appliedPaymentType, string description, decimal amount, bool raiseEvents = true);

        /// <summary>
        /// Saves an <see cref="IAppliedPayment"/>
        /// </summary>
        /// <param name="appliedPayment">The <see cref="IAppliedPayment"/> to be saved</param>
        /// <param name="raiseEvents">Optional boolean indicating whether or not to raise events</param>
        void Save(IAppliedPayment appliedPayment, bool raiseEvents = true);

        /// <summary>
        /// Saves a collection of <see cref="IAppliedPayment"/>
        /// </summary>
        /// <param name="appliedPayments">The collection of <see cref="IAppliedPayment"/>s to be saved</param>
        /// <param name="raiseEvents">Optional boolean indicating whether or not to raise events</param>
        void Save(IEnumerable<IAppliedPayment> appliedPayments, bool raiseEvents = true);

        /// <summary>
        /// Deletes a <see cref="IAppliedPayment"/>
        /// </summary>
        /// <param name="appliedPayment">The <see cref="IAppliedPayment"/> to be deleted</param>
        /// <param name="raiseEvents">Optional boolean indicating whether or not to raise events</param>
        void Delete(IAppliedPayment appliedPayment, bool raiseEvents = true);

        /// <summary>
        /// Deletes a collection of <see cref="IAppliedPayment"/>
        /// </summary>
        /// <param name="appliedPayments">The collection of <see cref="IAppliedPayment"/>s to be deleted</param>
        /// <param name="raiseEvents">Optional boolean indicating whether or not to raise events</param>
        void Delete(IEnumerable<IAppliedPayment> appliedPayments, bool raiseEvents = true);

        /// <summary>
        /// Returns a <see cref="IAppliedPayment"/> by it's unique 'key'
        /// </summary>
        /// <param name="key">The unique 'key' of the <see cref="IAppliedPayment"/></param>
        /// <returns>An <see cref="IAppliedPayment"/></returns>
        IAppliedPayment GetByKey(Guid key);

        /// <summary>
        /// Gets a collection of <see cref="IAppliedPayment"/>s by the payment key
        /// </summary>
        /// <param name="paymentKey">The payment key</param>
        /// <returns>A collection of <see cref="IAppliedPayment"/></returns>
        IEnumerable<IAppliedPayment> GetAppliedPaymentsByPaymentKey(Guid paymentKey);

        /// <summary>
        /// Gets a collection of <see cref="IAppliedPayment"/>s by the invoice key
        /// </summary>
        /// <param name="invoiceKey">The invoice key</param>
        /// <returns>A collection <see cref="IAppliedPayment"/></returns>
        IEnumerable<IAppliedPayment> GetAppliedPaymentsByInvoiceKey(Guid invoiceKey);
    }
}


text:
using LitJson;
using LinqToTwitter.Common;

namespace LinqToTwitter
{
    /// <summary>
    /// User returned by Control Stream query
    /// </summary>
    public class ControlStreamUser
    {
        public ControlStreamUser(JsonData userJson)
        {
            UserID = userJson.GetValue<ulong>("id");
            Name = userJson.GetValue<string>("name");
            DM = userJson.GetValue<bool>("dm");
        }

        /// <summary>
        /// User's unique Twitter ID
        /// </summary>
        public ulong UserID { get; set; }

        /// <summary>
        /// User's screen name
        /// </summary>
        public string Name { get; set; }

        /// <summary>
        /// Does the authenticated user have RW+DM access to user
        /// </summary>
        public bool DM { get; set; }
    }
}


text:
using content::SensorsListener;

namespace sensors {

ProviderImpl* ProviderImpl::GetInstance() {
  return Singleton<ProviderImpl>::get();
}

ProviderImpl::ProviderImpl()
    : listeners_(new ListenerList()) {
}

ProviderImpl::~ProviderImpl() {
}

void ProviderImpl::AddListener(SensorsListener* listener) {
  listeners_->AddObserver(listener);
}

void ProviderImpl::RemoveListener(SensorsListener* listener) {
  listeners_->RemoveObserver(listener);
}

void ProviderImpl::ScreenOrientationChanged(
    content::ScreenOrientation change) {
  listeners_->Notify(&SensorsListener::OnScreenOrientationChanged, change);
}

}  // namespace sensors


text:
_contributor @singhgarima_

For more information about fog [README](/README.md), or visit their website
[fog.io](www.fog.io).

## Vcloud API

Some useful links to get started on the vCloud API:

- [http://www.vmware.com/pdf/vcd_15_api_guide.pdf](http://www.vmware.com/pdf/vcd_15_api_guide.pdf)
- [vCloud API Programming Guide](http://pubs.vmware.com/vcd-51/index.jsp?topic=%2Fcom.vmware.vcloud.api.doc_51%2FGUID-86CA32C2-3753-49B2-A471-1CE460109ADB.html)

## Terminology

- Organization: An Organization is the fundamental vCloud Director grouping
  that contains users, the vApps that they create, and the resources the vApps
  use. It is a top-level container in a cloud that contains one or more
  Organization Virtual Data Centers (Org vDCs) and Catalog entities. It owns
  all the virtual resources for a cloud instance and can have many Org vDCs.[1]

- vApp: VMware vApp is a format for packaging and managing applications. A vApp
  can contain multiple virtual machines.[2]

- VM: A virtualized personal computer environment in which a guest
  operating system and associated application software can run. Multiple virtual
  machines can operate on the same managed host machine concurrently.[3]

- Catalogs & Catalog-Items: Catalog is used in organizations for storing content.
  Example: base images. Each item stored in catalog is referred as catalog item.

- vDC: Virtual Data Center. These are of two kinds provider vDCs (accessible to
  multiple organizations), and organization vDCs (accessible only by a given
  organization). In fog we refer to organization vDCs.

- Networks: You can setup various internal networks and assign various internal
  ip ranges to them

## What is the difference between a virtual appliance and a virtual machine?

A  virtual machine is a tightly isolated software container created to run on
virtualized platforms. It has four key virtualized resources (CPU, RAM,
Storage, and Networking); but requires the installation of an Operating System
and runs on one or more applications. A virtual appliance functions very much
like a virtual machine, possessing the four key characteristics of
compatibility, isolation, encapsulation, and hardware independence. However, a
virtual appliance contains a pre-installed, pre-configured Operating System
and an application stack optimized to provide a specific set of services.[3]

**References**

- [1] http://kb.vmware.com/selfservice/microsites/search.do?language=en_US&cmd=displayKC&externalId=1026316
- [2] http://www.vmware.com/pdf/vsphere4/r40/vsp_40_admin_guide.pdf
- [3] http://www.vmware.com/technical-resources/virtualization-topics/virtual-appliances/faqs


text:
PROG=	getresuid_suid

.include "../Makefile.cprog"


text:
"""Tests for fuzzer stats."""
# pylint: disable=protected-access
import ast
import datetime
import json
import os
import unittest

import flask
import mock
import webtest

from clusterfuzz._internal.datastore import data_types
from clusterfuzz._internal.google_cloud_utils import big_query
from clusterfuzz._internal.tests.test_libs import helpers as test_helpers
from clusterfuzz._internal.tests.test_libs import test_utils
from handlers import fuzzer_stats
from libs import helpers

DATA_DIRECTORY = os.path.join(os.path.dirname(__file__), 'fuzzer_stats_data')

def _read_data_file(data_file):
  """Helper function to read the contents of a data file."""
  with open(os.path.join(DATA_DIRECTORY, data_file)) as handle:
    return handle.read()

def _mock_query(fuzzer, jobs, group_by, date_start, date_end):
  return json.loads(
      _read_data_file('%s_%s_%s_%s_%s.txt' % (fuzzer, '_'.join(jobs)
                                              if jobs else None, group_by,
                                              date_start, date_end)))

@test_utils.with_cloud_emulators('datastore')
class TestBuildResults(unittest.TestCase):
  """Tests for fuzzer_stats build_results and wrappers."""

  def setUp(self):
    test_helpers.patch_environ(self)
    self.maxDiff = None  # pylint: disable=invalid-name

    data_types.Fuzzer(
        name='testFuzzer',
        stats_columns=('sum(t.blah) as blah, custom(j.new_crashes) '
                       'as new_crashes, _EDGE_COV as edge_coverage, '
                       '_FUNC_COV as func_coverage, '
                       '_CORPUS_SIZE as corpus_size, '
                       '_CORPUS_BACKUP as corpus_backup, '
                       '_QUARANTINE_SIZE as quarantine_size, '
                       '_COV_REPORT as coverage_report, '
                       '_FUZZER_RUN_LOGS as fuzzer_logs,'
                       '_PERFORMANCE_REPORT as performance_report'),
        stats_column_descriptions=(
            'blah: "blah description"\n'
            'func_coverage: "func coverage description"\n')).put()

    data_types.Fuzzer(
        name='testFuzzer2',
        stats_columns=('sum(t.blah) as blah, custom(j.new_crashes) '
                       'as new_crashes, _EDGE_COV as edge_coverage, '
                       '_FUNC_COV as func_coverage, '
                       '_CORPUS_SIZE as corpus_size, '
                       '_CORPUS_BACKUP as corpus_backup, '
                       '_QUARANTINE_SIZE as quarantine_size, '
                       '_COV_REPORT as coverage_report, '
                       '_FUZZER_RUN_LOGS as fuzzer_logs,'
                       '_PERFORMANCE_REPORT as performance_report'),
        stats_column_descriptions=(
            'blah: "blah description"\n'
            'func_coverage: "func coverage description"\n')).put()

    data_types.Job(
        name='job', environment_string='FUZZ_LOGS_BUCKET = bucket').put()

    now = datetime.datetime.utcnow()

    data_types.FuzzTarget(
        engine='testFuzzer', project='test-project', binary='1_fuzzer').put()
    data_types.FuzzTargetJob(
        fuzz_target_name='testFuzzer_1_fuzzer', job='job', last_run=now).put()

    data_types.FuzzTarget(
        engine='testFuzzer', project='test-project', binary='2_fuzzer').put()
    data_types.FuzzTargetJob(
        fuzz_target_name='testFuzzer_2_fuzzer', job='job', last_run=now).put()

    data_types.FuzzTarget(
        engine='testFuzzer', project='test-project', binary='3_fuzzer').put()
    data_types.FuzzTargetJob(
        fuzz_target_name='testFuzzer_3_fuzzer', job='job', last_run=now).put()

    data_types.FuzzTarget(
        engine='testFuzzer2', project='test-project', binary='1_fuzzer').put()
    data_types.FuzzTargetJob(
        fuzz_target_name='testFuzzer2_1_fuzzer', job='job', last_run=now).put()

    cov_info = data_types.CoverageInformation(
        fuzzer='2_fuzzer', date=datetime.date(2016, 10, 19))
    cov_info.edges_covered = 11
    cov_info.edges_total = 30
    cov_info.functions_covered = 10
    cov_info.functions_total = 15
    cov_info.html_report_url = 'https://report_for_2_fuzzer/20161019'
    cov_info.corpus_size_units = 20
    cov_info.corpus_size_bytes = 200
    cov_info.quarantine_size_units = 0
    cov_info.quarantine_size_bytes = 0
    cov_info.corpus_location = 'gs://corpus'
    cov_info.corpus_backup_location = 'gs://corpus-backup/file.zip'
    cov_info.quarantine_location = 'gs://quarantine'
    cov_info.put()

    cov_info = data_types.CoverageInformation(
        fuzzer='2_fuzzer', date=datetime.date(2016, 10, 21))
    cov_info.edges_covered = 15
    cov_info.edges_total = 30
    cov_info.functions_covered = 11
    cov_info.functions_total = 15
    cov_info.html_report_url = 'https://report_for_2_fuzzer/20161021'
    cov_info.corpus_size_units = 40
    cov_info.corpus_size_bytes = 400
    cov_info.quarantine_size_units = 8
    cov_info.quarantine_size_bytes = 80
    cov_info.corpus_location = 'gs://corpus'
    cov_info.corpus_backup_location = 'gs://corpus-backup/file.zip'
    cov_info.quarantine_location = 'gs://quarantine'
    cov_info.put()

    cov_info = data_types.CoverageInformation(
        fuzzer='1_fuzzer', date=datetime.date(2016, 10, 20))
    cov_info.edges_covered = 17
    cov_info.edges_total = 38
    cov_info.functions_covered = 12
    cov_info.functions_total = 19
    cov_info.html_report_url = 'https://report_for_1_fuzzer/20161020'
    cov_info.corpus_size_units = 47
    cov_info.corpus_size_bytes = 480
    cov_info.quarantine_size_units = 3
    cov_info.quarantine_size_bytes = 8
    cov_info.corpus_location = 'gs://corpus'
    cov_info.corpus_backup_location = 'gs://corpus-backup/file.zip'
    cov_info.quarantine_location = 'gs://quarantine'
    cov_info.put()

    self.client = mock.Mock(spec_set=big_query.Client)
    test_helpers.patch(self, [
        'clusterfuzz._internal.google_cloud_utils.big_query.Client',
    ])
    self.mock.Client.return_value = self.client

  def test_build_invalid_params(self):
    """Tests build_results with invalid/missing params."""
    with self.assertRaises(helpers.EarlyExitException):
      fuzzer_stats.build_results('', '', '', '', '')

  def test_build_by_fuzzer(self):
    """Tests basic build_results with valid parameters (group by fuzzer)."""
    build_args = [
        'testFuzzer', ['job'], 'by-fuzzer', '2016-10-20', '2016-10-21'
    ]
    self.client.raw_query.return_value = _mock_query(*build_args)
    result = fuzzer_stats.build_results(*build_args)

    self.assertDictEqual(
        ast.literal_eval(_read_data_file('by_fuzzer_expected.txt')), result)

  def test_build_by_day(self):
    """Tests basic build_results with valid parameters (group by day)."""
    build_args = [
        'testFuzzer_2_fuzzer', ['job'], 'by-day', '2016-10-19', '2016-10-21'
    ]
    self.client.raw_query.return_value = _mock_query(*build_args)
    result = fuzzer_stats.build_results(*build_args)

    self.assertDictEqual(
        ast.literal_eval(_read_data_file('by_day_expected.txt')), result)

  def test_build_by_job(self):
    """Tests basic build_results with valid parameters (group by job)."""
    build_args = [
        'testFuzzer_1_fuzzer', None, 'by-job', '2016-10-20', '2016-10-21'
    ]
    self.client.raw_query.return_value = _mock_query(*build_args)
    result = fuzzer_stats.build_results(*build_args)

    self.assertDictEqual(
        ast.literal_eval(_read_data_file('by_job_expected.txt')), result)

  def test_build_by_time(self):
    """Tests basic build_results with valid parameters (group by time)."""
    build_args = [
        'testFuzzer_2_fuzzer', ['job'], 'by-time', '2016-10-19', '2016-10-21'
    ]
    self.client.raw_query.return_value = _mock_query(*build_args)
    result = fuzzer_stats.build_results(*build_args)

    self.assertDictEqual(
        ast.literal_eval(_read_data_file('by_time_expected.txt')), result)

  def test_build_none_results(self):
    """Tests basic build_results, with bigquery returning None for some
    columns."""
    build_args = ['testFuzzer', None, 'by-fuzzer', '2016-10-20', '2016-10-22']
    self.client.raw_query.return_value = _mock_query(*build_args)
    result = fuzzer_stats.build_results(*build_args)

    self.assertDictEqual(
        ast.literal_eval(_read_data_file('by_fuzzer_expected_None.txt')),
        result)

  def test_build_old_results(self):
    """Tests build_results is caching old results appropriately."""
    test_helpers.patch(self, [
        'handlers.fuzzer_stats._build_old_results',
        'handlers.fuzzer_stats._build_todays_results',
        'clusterfuzz._internal.base.utils.utcnow',
    ])

    self.mock._build_old_results.return_value = None
    self.mock._build_todays_results.return_value = None
    self.mock.utcnow.return_value = datetime.datetime(2018, 5, 4)

    date_yesterday = '2018-05-03'
    build_args = [
        'testFuzzer_2_fuzzer', ['job'], 'by-time', date_yesterday,
        date_yesterday
    ]

    # Test that results from yesterday are cached for a full day.
    fuzzer_stats.build_results(*build_args)
    self.mock._build_old_results.assert_called_with(*build_args)
    self.assertEqual(0, self.mock._build_todays_results.call_count)

  def test_build_today_results(self):
    """Tests build_results is caching results from today appropriately."""
    test_helpers.patch(self, [
        'handlers.fuzzer_stats._build_old_results',
        'handlers.fuzzer_stats._build_todays_results',
        'clusterfuzz._internal.base.utils.utcnow',
    ])

    self.mock._build_old_results.return_value = None
    self.mock._build_todays_results.return_value = None
    self.mock.utcnow.return_value = datetime.datetime(2018, 5, 4)

    date_today = '2018-05-04'
    build_args = [
        'testFuzzer_2_fuzzer', ['job'], 'by-time', date_today, date_today
    ]

    # Test that results from yesterday are cached for a full day.
    fuzzer_stats.build_results(*build_args)
    self.mock._build_todays_results.assert_called_with(*build_args)
    self.assertEqual(0, self.mock._build_old_results.call_count)

@test_utils.with_cloud_emulators('datastore')
class TestPermissions(unittest.TestCase):
  """Permissions tests."""

  def setUp(self):
    test_helpers.patch(self, [
        'libs.auth.get_current_user',
        'libs.auth.is_current_user_admin',
        'handlers.fuzzer_stats.build_results',
    ])

    self.mock.build_results.return_value = json.dumps({})

    flaskapp = flask.Flask('testflask')
    flaskapp.add_url_rule(
        '/fuzzer-stats/load',
        view_func=fuzzer_stats.LoadHandler.as_view('/fuzzer-stats/load'))
    self.app = webtest.TestApp(flaskapp)

    data_types.ExternalUserPermission(
        email='test@user.com',
        entity_kind=data_types.PermissionEntityKind.JOB,
        entity_name='job1',
        is_prefix=False,
        auto_cc=data_types.AutoCCType.ALL).put()

    data_types.ExternalUserPermission(
        email='test@user.com',
        entity_kind=data_types.PermissionEntityKind.JOB,
        entity_name='job2',
        is_prefix=False,
        auto_cc=data_types.AutoCCType.ALL).put()

    data_types.Job(name='job1').put()
    data_types.Job(name='job2').put()
    data_types.Job(name='job3').put()

  def test_internal_user_with_job(self):
    """Test internal user access (with job)."""
    self.mock.is_current_user_admin.return_value = True
    response = self.app.post_json(
        '/fuzzer-stats/load', {
            'fuzzer': 'fuzzer',
            'job': 'job1',
            'group_by': 'by-fuzzer',
            'date_start': '2017-01-01',
            'date_end': '2017-01-07',
        })
    self.assertEqual(200, response.status_int)
    self.mock.build_results.assert_called_with('fuzzer', ['job1'], 'by-fuzzer',
                                               '2017-01-01', '2017-01-07')

  def test_internal_user_without_job(self):
    """Test internal user access (with job)."""
    self.mock.is_current_user_admin.return_value = True
    response = self.app.post_json(
        '/fuzzer-stats/load', {
            'fuzzer': 'fuzzer',
            'group_by': 'by-fuzzer',
            'date_start': '2017-01-01',
            'date_end': '2017-01-07',
        })
    self.assertEqual(200, response.status_int)
    self.mock.build_results.assert_called_with('fuzzer', None, 'by-fuzzer',
                                               '2017-01-01', '2017-01-07')

  def test_external_user_with_job(self):
    """Test external user access (job specified)."""
    self.mock.is_current_user_admin.return_value = False
    self.mock.get_current_user().email = 'test@user.com'

    response = self.app.post_json(
        '/fuzzer-stats/load', {
            'fuzzer': 'fuzzer',
            'job': 'job1',
            'group_by': 'by-fuzzer',
            'date_start': '2017-01-01',
            'date_end': '2017-01-07',
        })
    self.assertEqual(200, response.status_int)

    response = self.app.post_json(
        '/fuzzer-stats/load', {
            'fuzzer': 'fuzzer',
            'job': 'job2',
            'group_by': 'by-fuzzer',
            'date_start': '2017-01-01',
            'date_end': '2017-01-07',
        })
    self.assertEqual(200, response.status_int)

    response = self.app.post_json(
        '/fuzzer-stats/load', {
            'fuzzer': 'fuzzer',
            'job': 'job3',
            'group_by': 'by-fuzzer',
            'date_start': '2017-01-01',
            'date_end': '2017-01-07',
        },
        expect_errors=True)
    self.assertEqual(403, response.status_int)

  def test_external_user_without_job(self):
    """Test external user access (no job specified)."""
    self.mock.is_current_user_admin.return_value = False
    self.mock.get_current_user().email = 'test@user.com'

    response = self.app.post_json(
        '/fuzzer-stats/load', {
            'fuzzer': 'fuzzer',
            'group_by': 'by-fuzzer',
            'date_start': '2017-01-01',
            'date_end': '2017-01-07',
        })
    self.assertEqual(200, response.status_int)
    self.mock.build_results.assert_called_with(
        'fuzzer', ['job1', 'job2'], 'by-fuzzer', '2017-01-01', '2017-01-07')

class TestGetDate(unittest.TestCase):
  """Test for fuzzer_stats._get_date."""

  def setUp(self):
    test_helpers.patch(self, [
        'clusterfuzz._internal.base.utils.utcnow',
    ])

  def test_get_date_empty_date(self):
    """Test _get_date correctly replaces an empty date."""
    self.mock.utcnow.return_value = datetime.datetime(2018, 5, 4)
    # Test that when given '' and 1, it returns the string representing
    # the date one day ago.
    self.assertEqual(fuzzer_stats._get_date('', 1), '2018-05-03')

  def test_get_date_non_empty_date(self):
    """Test _get_date just returns a non-empty date."""
    date = '2018-5-1'
    self.assertEqual(fuzzer_stats._get_date(date, 1), date)


text:
'use strict';

angular.module('core.student').factory('Student', ['$resource',
    function ($resource) {
        return $resource('/api/students/:id', {id: '@id'},
            {
                'update': {method: 'PUT'}
            });
    }]);


text:
process.env.TZ = 'UTC'
var expect = require('chai').expect
var ini_set = require('../../../../src/php/info/ini_set') // eslint-disable-line no-unused-vars,camelcase
var ini_get = require('../../../../src/php/info/ini_get') // eslint-disable-line no-unused-vars,camelcase
var ini_get = require('../../../../src/php/info/ini_get.js') // eslint-disable-line no-unused-vars,camelcase

describe('src/php/info/ini_get.js (tested in test/languages/php/info/test-ini_get.js)', function () {
  it('should pass example 1', function (done) {
    var expected = 'Asia/Hong_Kong'
    ini_set('date.timezone', 'Asia/Hong_Kong')
    var result = ini_get('date.timezone')
    expect(result).to.deep.equal(expected)
    done()
  })
})


text:
package edu.fudan.eservice.common.utils;

import java.util.Arrays;
import java.util.Properties;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.mongodb.DB;
import com.mongodb.DBCollection;
import com.mongodb.MongoClient;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
/**
 * 获取MongoDB的操作对象,MongoDB的Java Driver自带连接池,此单件对象用于返回操作用的MongoClient或者认证过的DB对象
 * <pre>
 *  //获取MongoClient对象
 * 	MongoClient client=MongoUtil.getInstance().getClient();
 *  //获取配置文件config.properties中默认的mongo.db并使用mongo.user/mongo.pwd认证
 *  DB db=MongoUtil.getInstance().getDB();
 * @author wking
 *
 */
public class MongoUtil {

	private static MongoUtil mu=null;
	private static Log log=LogFactory.getLog(MongoUtil.class);
	private MongoClient client;
	private Properties props;
	protected MongoUtil() {
		props=new Properties();
		try {
			props.load(this.getClass().getResourceAsStream("/dbconnect.properties"));
			String user=props.getProperty("mongo.user");
			if(!CommonUtil.isEmpty(user))
				client=new MongoClient(new ServerAddress(props.getProperty("mongo.ip","localhost"),Integer.parseInt(props.getProperty("mongo.port","27017"))),
						Arrays.asList(MongoCredential.createCredential(user, props.getProperty("mongo.db"), props.getProperty("mongo.pwd").toCharArray())));

			else
			client=new MongoClient(new ServerAddress(String.valueOf(props.get("mongo.ip")),Integer.parseInt(String.valueOf(props.get("mongo.port")))));

		} catch (Exception e) {
			log.error("MongoDB Connect Fail");
		}

	}
	/**
	 * 单件模式获取MongoUtil实例
	 * @return
	 */

	public static synchronized MongoUtil getInstance()
	{
		if(mu==null)
		{
			mu=new MongoUtil();
		}
		return mu;
	}
	/**
	 * 获取MongoClient
	 * @return
	 */
	public MongoClient getClient()
	{
		return client;
	}
	/**
	 * 获取默认的DB对象并完成认证
	 * @return
	 */
	public DB getDB()
	{

		DB ret= client.getDB(String.valueOf(props.get("mongo.db")));
			return ret;
	}

	/**
	 *
	* @Title: getCollection
	* @Description: 获取一个集合
	* @param collectionName
	* @return
	 */
	public DBCollection getCollection(String collectionName) {
		return getDB().getCollection(collectionName);
	}
}


text:
opal_filter "Singleton" do
  fails "Singleton.instance returns an instance of the singleton's clone"
  fails "Singleton.instance returns the same instance for multiple class to instance on clones"

  fails "Singleton._load returns the singleton instance for anything passed in"
  fails "Singleton._load returns the singleton instance for anything passed in to subclass"
end