text:
from __future__ import unicode_literals
from __future__ import absolute_import

import copy
import datetime
import os
import sys
import threading
import time
from abc import ABCMeta, abstractmethod
import multiprocessing.managers

if False:
    from typing import Dict
    from typing import Any
    from typing import List
    from typing import Optional

from scalyr_agent import scalyr_logging as scalyr_logging, StoppableThread
from scalyr_agent.agent_status import CopyingManagerWorkerSessionStatus
from scalyr_agent.log_processing import LogFileProcessor
from scalyr_agent.util import RateLimiter
from scalyr_agent import util as scalyr_util
from scalyr_agent.configuration import Configuration
from scalyr_agent.scalyr_client import (
    create_client,
    create_new_client,
    ScalyrClientSession,
    ScalyrClientSessionStatus,
)
from scalyr_agent.copying_manager.common import write_checkpoint_state_to_file

import six

log = scalyr_logging.getLogger(__name__)

WORKER_SESSION_CHECKPOINT_FILENAME_PREFIX = "checkpoints-worker-"
WORKER_SESSION_CHECKPOINT_FILENAME_GLOB = "{0}*-*.json".format(
    WORKER_SESSION_CHECKPOINT_FILENAME_PREFIX
)

class CopyingParameters(object):
    """Tracks the copying parameters that should be used for sending requests to Scalyr and adjusts them over time
    according to success and failures of requests.

    The copying parameters boil down to two parameters:  the maximum number of bytes that can be sent to Scalyr
    in a request (self.current_bytes_allowed_to_send), and the minimum amount of time to wait between requests
    (self.current_sleep_interval).

    This implements a truncated binary backoff algorithm.
    """

    def __init__(self, configuration):
        """Initialize the parameters based on the thresholds defined in the configuration file.

        @param configuration: The configuration file.
        @type configuration: configuration.Configuration
        """
        # The current maximum number of bytes that can be sent in a single request to Scalyr.
        # This will be adjusted over time.
        self.current_bytes_allowed_to_send = configuration.max_allowed_request_size
        # The current time to sleep between requests to Scalyr.
        # This will be adjusted over time.
        self.current_sleep_interval = configuration.max_request_spacing_interval

        # The maximum value current_bytes_allowed_to_send can take.
        self.__max_allowed_request_size = configuration.max_allowed_request_size
        # The minimum value current_bytes_allowed_to_send can take.
        self.__min_allowed_request_size = configuration.min_allowed_request_size

        # The maximum value current_sleep_interval can take.
        self.__max_request_spacing_interval = configuration.max_request_spacing_interval
        # The minimum value current_sleep_interval can take.
        self.__min_request_spacing_interval = configuration.min_request_spacing_interval

        # The maximum value current_sleep_interval can take when there has been an error.
        self.__max_error_request_spacing_interval = (
            configuration.max_error_request_spacing_interval
        )

        # The low water mark for the number of bytes sent in a request.  If, when we go to collect the lines that
        # need to be sent from all logs, the amount is lower than this, then we will sleep more between requests.
        self.__low_water_bytes_sent = configuration.low_water_bytes_sent
        # The percentage to adjust the sleeping time if the lower_water_bytes_sent mark is not met.
        self.__low_water_request_spacing_adjustment = (
            configuration.low_water_request_spacing_adjustment
        )
        # The high water mark for the number of bytes sent in a request.  If, when we go to collect the lines that
        # need to be sent from all logs, the amount is higher than this, then we will sleep less between requests.
        self.__high_water_bytes_sent = configuration.high_water_bytes_sent
        # The percentage to adjust the sleeping time if the high_water_bytes_sent mark is exceeded.
        self.__high_water_request_spacing_adjustment = (
            configuration.high_water_request_spacing_adjustment
        )

        # The percentage to adjust the sleeping time if the last request was a failure.
        self.__failure_request_spacing_adjustment = (
            configuration.failure_request_spacing_adjustment
        )

        # The percentage to adjust the allowed request size if we get a back a 'requestTooLarge' response from
        # the server.
        self.__request_too_large_adjustment = configuration.request_too_large_adjustment

    def update_params(self, result, bytes_sent):
        """Updates the current_bytes_allowed_to_send and current_sleep_interval based on the result from the last
        request as well as the number of bytes sent.

        @param result: The status field from the response from the server for the last request.
        @param bytes_sent: The number of bytes sent in the last request.
        @type result: str
        @type bytes_sent: int
        """
        # The algorithm is as follows:
        #    For the sleep time:  If it is a success, then we multiple the current sleep time by either the
        #      lower or high water spacing adjustment, depending if we sent fewer than the lower water bytes or
        #      more than the high water bytes.  (If we sent a number between them, then no adjustment is made).
        #      If the request was a failure, we adjust it by the failure request failure adjustment.
        #   For the size, we only adjust the request downwards if we get a 'requestTooLarge' message from the server,
        #      but on success, we put the allowed size back up to the maximum.
        if result == "success":
            max_request_spacing_interval = self.__max_request_spacing_interval
            if bytes_sent < self.__low_water_bytes_sent:
                self.current_sleep_interval *= (
                    self.__low_water_request_spacing_adjustment
                )
            elif bytes_sent > self.__high_water_bytes_sent:
                self.current_sleep_interval *= (
                    self.__high_water_request_spacing_adjustment
                )
        else:
            self.current_sleep_interval *= self.__failure_request_spacing_adjustment
            max_request_spacing_interval = self.__max_error_request_spacing_interval

        if result == "success":
            self.current_bytes_allowed_to_send = self.__max_allowed_request_size
        elif "requestTooLarge" in result:
            self.current_bytes_allowed_to_send = int(
                bytes_sent * self.__request_too_large_adjustment
            )

        self.current_bytes_allowed_to_send = self.__ensure_within(
            self.current_bytes_allowed_to_send,
            self.__min_allowed_request_size,
            self.__max_allowed_request_size,
        )

        self.current_sleep_interval = self.__ensure_within(
            self.current_sleep_interval,
            self.__min_request_spacing_interval,
            max_request_spacing_interval,
        )

    def __ensure_within(self, value, min_value, max_value):
        """Return value subject to the constraints that it must be greater than min_value and less than max_value.

        @param value: The raw value.
        @param min_value: The minimum allowed value.
        @param max_value: The maximum allowed value.

        @type value: float
        @type min_value: float
        @type max_value: float

        @return: The value to use.  If it is between min_value and max_value, then the original value.  Otherwise,
            either min_value or max_value depending on which constraint was violated.
        @rtype: float
        """
        if value < min_value:
            value = min_value
        elif value > max_value:
            value = max_value
        return value

class AddEventsTask(object):
    """Encapsulates the state for a pending AddEventRequest."""

    def __init__(self, add_events_request, completion_callback):
        """Initializes the instance.

        @param add_events_request: The AddEventRequest to send using a ScalyrClientSession
        @param completion_callback:  The completion callback to invoke once the request has been successfully sent.
            It takes a single int argument, which must be one of the three values:  LogFileProcessor.SUCCESS,
            LogFileProcessor.FAIL_AND_RETRY, LogFileProcessor.FAIL_AND_DROP.
        @type add_events_request: scalyr_client.AddEventRequest
        @type completion_callback: function that takes an int argument
        """
        # The pending AddEventsRequest to send.
        self.add_events_request = add_events_request
        # The calllback to invoke once the request has completed.
        self.completion_callback = completion_callback
        # If there is a AddEventsTask object already created for the next request due to pipelining, this is set to it.
        # This must be the next request if this request is successful, otherwise, we will lose bytes.
        self.next_pipelined_task = None

class CopyingManagerWorkerSessionInterface(six.with_metaclass(ABCMeta)):
    """
    The interface for the Copying manager worker session classes.
    """

    @abstractmethod
    def start_worker_session(self):
        pass

    @abstractmethod
    def stop_worker_session(self, wait_on_join=True, join_timeout=5):
        """Stops the worker session.

        @param wait_on_join: If True, will block on a join of the thread running the worker session.
        @param join_timeout: The maximum number of seconds to block for the join.
        """
        pass

    @abstractmethod
    def generate_status(
        self, warn_on_rate_limit=False
    ):  # type: (bool) -> CopyingManagerWorkerSessionStatus
        """Generate the status for the copying manager worker session to be reported.

        This is used in such features as 'scalyr-agent-2 status -v'.

        Note, this method is thread safe.  It needs to be since another thread will ask this object for its
        status.

        @return:  The status object containing the statistics for the copying manager worker session.
        """
        pass

    @abstractmethod
    def wait_for_copying_to_begin(self):
        """
        Block the current thread until session's instance has begun copying.
        """
        pass

    @abstractmethod
    def create_and_schedule_new_log_processor(self, *args, **kwargs):
        """
        Creates and also schedules a new log processor. It will be added to the main collection only on the next iteration,
        so we don't have to guard the main collection with lock.
        """
        pass

    @abstractmethod
    def augment_scalyr_client_user_agent(
        self, fragments
    ):  # type: (List[six.text_type]) -> None
        """
        Modifies User-Agent header of the Scalyr client session.

        @param fragments String fragments to append (in order) to the standard user agent data
        """
        pass

    @abstractmethod
    def get_id(self):
        pass

    @abstractmethod
    def generate_scalyr_client_status(self):  # type: () -> ScalyrClientSessionStatus
        """
        Get the status object of the Scalyr client.
        """
        pass

    @abstractmethod
    def log_worker_session_status(self):
        """
        Write the main information about the worker and session stats to the worker's agent log.
        """
        pass

class CopyingManagerWorkerSession(
    StoppableThread, CopyingManagerWorkerSessionInterface
):
    """
    Abstraction which is responsible for copying the log files to the Scalyr servers.

    This is run as its own thread.
    """

    def __init__(self, configuration, worker_config_entry, session_id, is_daemon=False):
        # type: (Configuration, Dict, six.text_type, bool) -> None
        """Initializes the copying manager worker session.

        @param configuration: The configuration file containing which log files need to be copied listed in the
            configuration file.
        @param worker_config_entry:
        @param session_id: Id of the worker session.
        @:param is_daemon: If true, start a session thread as a daemon thread.

        """
        StoppableThread.__init__(
            self,
            name="copying manager worker session thread #%s" % session_id,
            is_daemon=is_daemon,
        )

        self._id = six.text_type(session_id)
        self.__config = configuration
        self.__worker_config_entry = worker_config_entry

        self.__new_scalyr_client = None

        # Rate limiter
        self.__rate_limiter = None
        if self.__config.parsed_max_send_rate_enforcement:
            self.__rate_limiter = RateLimiter(
                # Making the bucket size 4 times the bytes per second, it shouldn't affect the rate over a long time
                # meaningfully but we don't have a rationale for this value.
                # TODO: Make this configurable as part of the `max_send_rate_enforcement` configuration option
                self.__config.parsed_max_send_rate_enforcement * 4.0,
                self.__config.parsed_max_send_rate_enforcement,
            )

        # collect monitor-specific extra server-attributes.  seed with a copy of the attributes and converted to a dict.
        self.__expanded_server_attributes = self.__config.server_attributes.to_dict()

        # The dict of LogFileProcessors that are processing the lines from matched log files.
        self.__log_processors = []  # type: List[LogFileProcessor]

        # Temporary collection of recently added log processors.
        # Every log processor which is added during the iteration of the worker session is placed in here.
        # All those log processors will be added to the main collection
        # on the beginning of the next iteration.
        # It stores all new log processors before they are added to the main log processors list.
        self.__new_log_processors = []  # type: List[LogFileProcessor]

        # A lock that protects the status variables and the __log_processors variable, the only variables that
        # are access in generate_status() which needs to be thread safe.
        self.__lock = threading.Lock()

        # The current pending AddEventsTask.  We will retry the contained AddEventsRequest several times.
        self.__pending_add_events_task = None

        # The next LogFileProcessor that should have log lines read from it for transmission.
        self.__current_processor = 0

        # The client to use for sending the data.  Set in the start_manager call.
        self.__scalyr_client = None  # type: Optional[ScalyrClientSession]
        # The last time we scanned for new files that match the __log_matchers.
        self.__last_new_file_scan_time = 0

        # last time when worker logged its stats.
        self.__last_stats_message_time = 0

        # Status variables that track statistics reported to the status page.
        self._last_attempt_time = None
        self.__last_success_time = None
        self.__last_attempt_size = None
        self.__last_response = None
        self.__last_response_status = None
        self.__total_bytes_uploaded = 0
        self.__total_errors = 0
        self.__total_rate_limited_time = 0
        self.__rate_limited_time_since_last_status = 0

        # A semaphore that we increment when this object has begun copying files (after first scan).
        self.__copying_semaphore = threading.Semaphore(0)

        # debug leaks
        self.__disable_scan_for_new_bytes = configuration.disable_scan_for_new_bytes

        # Statistics for copying_manager_worker_status messages
        self.total_copy_iterations = 0
        self.total_read_time = 0
        self.total_waiting_time = 0
        self.total_blocking_response_time = 0
        self.total_request_time = 0
        self.total_pipelined_requests = 0

        self.__last_total_bytes_skipped = 0
        self.__last_total_bytes_copied = 0
        self.__last_total_bytes_pending = 0

    @property
    def expanded_server_attributes(self):
        """Return deepcopy of expanded server attributes"""
        return copy.deepcopy(self.__expanded_server_attributes)

    def run(self):
        """Processes the log files, which were added by copying manager, reading their
        bytes, applying redaction and sampling rules, and then sending them to the server.

        This method will not terminate until the thread has been stopped.
        """

        # So the scanning.. every scan:
        #   - Update the file length counts of all current scanners:
        #   - Then pick up where you left off, getting X bytes as determined that abstraction
        #   - Send it to the client
        #   - determine success or not.. if success, update it.
        #   - sleep

        if self.__config.copying_thread_profile_interval > 0:
            import cProfile

            profiler = cProfile.Profile()
            profiler.enable()
            profile_dump_interval = self.__config.copying_thread_profile_interval
        else:
            profiler = None
            profile_dump_interval = 0

        current_time = time.time()

        try:
            # noinspection PyBroadException
            try:
                # The copying params that tell us how much we are allowed to send and how long we have to wait between
                # attempts.
                copying_params = CopyingParameters(self.__config)

                current_time = time.time()

                # Just initialize the last time we had a success to now.  Make the logic below easier.
                # NOTE: We set this variable to current (start time) even if we never successfuly
                # establish a connection because we want eventually drop __pending_add_events_task
                # even if we can't establish a connection. If we didn't do that, that queue could
                # grow unbounded.
                # Because of that, we need to take this behavior into account when updating
                # "__last_success_time" variable which is used for status reporting. We do that by
                # utilizing another last_success_status variable which only gets updated when we
                # successfuly send the request to the server.
                last_success = current_time
                last_success_status = None

                # Force the agent to write a new full checkpoint as soon as it can
                last_full_checkpoint_write = 0.0

                pipeline_byte_threshold = self.__config.pipeline_threshold * float(
                    self.__config.max_allowed_request_size
                )

                log.info(
                    "Copying manager worker session #%s started. Pid: '%s'"
                    % (self._id, os.getpid())
                )

                # Create new Scalyr client
                self._init_scalyr_client()

                # add one time interval to skip the status logging immediately after start,
                last_status_message_time = (
                    current_time
                    + self.__config.default_worker_session_status_message_interval
                )

                # We are about to start copying.  We can tell waiting threads.
                self.__copying_semaphore.release()
                while self._run_state.is_running():
                    log.log(
                        scalyr_logging.DEBUG_LEVEL_1,
                        "At top of copy log files loop on worker session #%s. (Iteration #%s)"
                        % (self._id, self.total_copy_iterations),
                    )
                    current_time = time.time()
                    pipeline_time = 0.0
                    # noinspection PyBroadException
                    try:
                        # If we have a pending request and it's been too taken too long to send it, just drop it
                        # on the ground and advance.
                        if current_time - last_success > self.__config.max_retry_time:
                            if self.__pending_add_events_task is not None:
                                self.__pending_add_events_task.completion_callback(
                                    LogFileProcessor.FAIL_AND_DROP
                                )
                                self.__pending_add_events_task = None
                            # Tell all of the processors to go to the end of the current log file.  We will start
                            # copying
                            # from there.
                            for processor in self.__log_processors:
                                processor.skip_to_end(
                                    "Too long since last successful request to server.",
                                    "skipNoServerSuccess",
                                    current_time=current_time,
                                )

                        self.__add_new_log_processors()

                        # Collect log lines to send if we don't have one already.
                        if self.__pending_add_events_task is None:
                            self.__pending_add_events_task = self.__get_next_add_events_task(
                                copying_params.current_bytes_allowed_to_send
                            )
                        else:
                            log.log(
                                scalyr_logging.DEBUG_LEVEL_1,
                                "Have pending batch of events, retrying to send.",
                            )
                            # Take a look at the file system and see if there are any new bytes pending.  This updates
                            # the statistics for each pending file.  This is important to do for status purposes if we
                            # have not tried to invoke get_next_send_events_task in a while (since that already updates
                            # the statistics).
                            self.__scan_for_new_bytes(current_time=current_time)

                        # Try to send the request if we have one.
                        if self.__pending_add_events_task is not None:
                            log.log(
                                scalyr_logging.DEBUG_LEVEL_1,
                                "Sending an add event request",
                            )
                            # Send the request, but don't block for the response yet.
                            send_request_time_start = time.time()
                            get_response = None
                            if not self.__config.use_new_ingestion:
                                get_response = self._send_events(
                                    self.__pending_add_events_task
                                )

                            # Check to see if pipelining should be disabled
                            # TODO: uncomment this when the pipelining is enabled again.
                            # disable_pipelining = (
                            #     self.__has_pending_log_changes()
                            # )

                            # If we are sending very large requests, we will try to optimize for future requests
                            # by overlapping building the request with waiting for the response on the current request
                            # (pipelining).

                            if (
                                self.__pending_add_events_task.add_events_request.current_size
                                >= pipeline_byte_threshold
                                and self.__pending_add_events_task.next_pipelined_task
                                is None
                                # TODO: pipelining is temporarily disabled and should be enabled back after
                                #  other issues related to the transitioning to the sharded copying manager are solved..
                                # and not disable_pipelining
                            ):

                                # Time how long it takes us to build it because we will subtract it from how long we
                                # have to wait before we send the next request.
                                pipeline_time = time.time()
                                self.total_pipelined_requests += 1
                                self.__pending_add_events_task.next_pipelined_task = self.__get_next_add_events_task(
                                    copying_params.current_bytes_allowed_to_send,
                                    for_pipelining=True,
                                )
                            else:
                                pipeline_time = 0.0

                            # Now block for the response.
                            blocking_response_time_start = time.time()
                            if self.__config.use_new_ingestion:
                                result = "success"
                                bytes_sent = 0
                                full_response = ""
                            else:
                                (result, bytes_sent, full_response) = get_response()
                            blocking_response_time_end = time.time()
                            self.total_blocking_response_time += (
                                blocking_response_time_end
                                - blocking_response_time_start
                            )
                            self.total_request_time += (
                                blocking_response_time_end - send_request_time_start
                            )

                            if pipeline_time > 0:
                                pipeline_time = time.time() - pipeline_time
                            else:
                                pipeline_time = 0.0

                            log.log(
                                scalyr_logging.DEBUG_LEVEL_1,
                                'Sent %ld bytes and received response with status="%s".',
                                bytes_sent,
                                result,
                            )

                            if result != "success":
                                # Log the whole response body in case server returns non-success
                                # response so we can troubleshoot the error (aka is it bug in the
                                # agent code or similar)
                                log.log(
                                    scalyr_logging.DEBUG_LEVEL_5,
                                    'Received server response with status="%s" and body: %s',
                                    result,
                                    full_response,
                                )

                            log_bytes_sent = 0
                            if (
                                result == "success"
                                or "discardBuffer" in result
                                or "requestTooLarge" in result
                            ):
                                next_add_events_task = None
                                try:
                                    if result == "success":
                                        log_bytes_sent = self.__pending_add_events_task.completion_callback(
                                            LogFileProcessor.SUCCESS
                                        )
                                        next_add_events_task = (
                                            self.__pending_add_events_task.next_pipelined_task
                                        )
                                    elif "discardBuffer" in result:
                                        self.__pending_add_events_task.completion_callback(
                                            LogFileProcessor.FAIL_AND_DROP
                                        )
                                    else:
                                        self.__pending_add_events_task.completion_callback(
                                            LogFileProcessor.FAIL_AND_RETRY
                                        )
                                finally:
                                    # No matter what, we want to throw away the current event since the server said we
                                    # could.  We have seen some bugs where we did not throw away the request because
                                    # an exception was thrown during the callback.
                                    self.__pending_add_events_task = (
                                        next_add_events_task
                                    )
                                    self.__write_active_checkpoint_state(current_time)

                            if result == "success":
                                last_success = current_time
                                last_success_status = current_time

                            # Rate limit based on amount of copied log bytes in a successful request
                            if self.__rate_limiter:
                                time_slept = self.__rate_limiter.block_until_charge_succeeds(
                                    log_bytes_sent
                                )
                                self.__total_rate_limited_time += time_slept
                                self.__rate_limited_time_since_last_status += time_slept
                                self.total_waiting_time += time_slept

                        else:
                            result = "failedReadingLogs"
                            bytes_sent = 0
                            full_response = ""

                            log.error("Failed to read logs for copying.  Will re-try")

                        # Update the statistics and our copying parameters.
                        self.__lock.acquire()
                        copying_params.update_params(result, bytes_sent)
                        self._last_attempt_time = current_time
                        self.__last_success_time = last_success_status
                        self.__last_attempt_size = bytes_sent
                        self.__last_response = six.ensure_text(full_response)
                        self.__last_response_status = result
                        if result == "success":
                            self.__total_bytes_uploaded += bytes_sent
                        self.__lock.release()

                        # if worker session is in another process than write its main stats periodically to the its log.
                        if (
                            self.__config.use_multiprocess_workers
                            and last_status_message_time
                            + self.__config.default_worker_session_status_message_interval
                            < current_time
                        ):
                            self.log_worker_session_status()
                            last_status_message_time = current_time

                        if profiler is not None:
                            seconds_past_epoch = int(time.time())
                            if seconds_past_epoch % profile_dump_interval == 0:
                                profiler.disable()
                                profiler.dump_stats(
                                    "%s%s%s"
                                    % (
                                        self.__config.copying_thread_profile_output_path,
                                        "copying_worker_session_",
                                        datetime.datetime.utcnow().strftime(
                                            "%H_%M_%S.out"
                                        ),
                                    )
                                )
                                profiler.enable()
                    except Exception:
                        # TODO: Do not catch Exception here.  That is too broad.  Disabling warning for now.
                        log.exception(
                            "Failed while attempting to scan and transmit logs"
                        )
                        log.log(
                            scalyr_logging.DEBUG_LEVEL_1,
                            "Failed while attempting to scan and transmit logs",
                        )
                        self.__lock.acquire()
                        self._last_attempt_time = current_time
                        self.__total_errors += 1
                        self.__lock.release()

                    if (
                        current_time - last_full_checkpoint_write
                        > self.__config.full_checkpoint_interval
                    ):
                        self._write_full_checkpoint_state(current_time)
                        last_full_checkpoint_write = current_time

                    if pipeline_time < copying_params.current_sleep_interval:
                        self._sleep_but_awaken_if_stopped(
                            copying_params.current_sleep_interval - pipeline_time
                        )
                        self.total_waiting_time += (
                            copying_params.current_sleep_interval - pipeline_time
                        )

                    # End of the copy loop
                    self.total_copy_iterations += 1
            except Exception:
                # If we got an exception here, it is caused by a bug in the program.
                log.exception("Log copying failed due to exception")
                # if error has occurred earlier than this semaphore is released,
                # than we need to release it to be sure that the copying manager won't stuck in deadlock.
                self.__copying_semaphore.release()

                # stop worker session's thread.
                sys.exit(1)
        finally:
            self._write_full_checkpoint_state(current_time)
            for processor in self.__log_processors:
                processor.close()

            if self.__scalyr_client is not None:
                self.__scalyr_client.close()
            if self.__new_scalyr_client is not None:
                self.__new_scalyr_client.close()

            if profiler is not None:
                profiler.disable()

            log.log(
                scalyr_logging.DEBUG_LEVEL_0,
                "Worker session '%s' is finished." % self._id,
            )

    def wait_for_copying_to_begin(self):
        """
        Block the current thread until worker session's instance has begun copying.
        """
        self.__copying_semaphore.acquire(True)
        return

    def generate_status(self, warn_on_rate_limit=False):
        # type: (bool) -> CopyingManagerWorkerSessionStatus
        try:
            self.__lock.acquire()

            result = CopyingManagerWorkerSessionStatus()
            result.session_id = self._id
            result.pid = os.getpid()
            result.total_bytes_uploaded = self.__total_bytes_uploaded
            result.last_success_time = self.__last_success_time
            result.last_attempt_time = self._last_attempt_time
            result.last_attempt_size = self.__last_attempt_size
            result.last_response = self.__last_response
            result.last_response_status = self.__last_response_status
            result.total_errors = self.__total_errors
            result.total_rate_limited_time = self.__total_rate_limited_time
            result.rate_limited_time_since_last_status = (
                self.__rate_limited_time_since_last_status
            )
            result.total_copy_iterations = self.total_copy_iterations
            result.total_read_time = self.total_read_time
            result.total_waiting_time = self.total_waiting_time
            result.total_blocking_response_time = self.total_blocking_response_time
            result.total_request_time = self.total_request_time
            result.total_pipelined_requests = self.total_pipelined_requests

            for processor in self.__log_processors:
                log_processor_status = processor.generate_status()
                result.log_processors.append(log_processor_status)

            if self._last_attempt_time:
                result.health_check_result = "Good"
                if (
                    time.time()
                    > self._last_attempt_time
                    + self.__config.healthy_max_time_since_last_copy_attempt
                ):
                    result.health_check_result = (
                        "Worker session '%s' failed, max time since last copy attempt (%s seconds) exceeded"
                        % (
                            self._id,
                            self.__config.healthy_max_time_since_last_copy_attempt,
                        )
                    )

        finally:
            self.__lock.release()
        if warn_on_rate_limit:
            if self.__rate_limited_time_since_last_status > 0:
                log.warning(
                    "Warning, the maximum send rate has been exceeded (configured through "
                    "'max_send_rate_enforcement').  Log upload is being delayed and may result in skipped log "
                    "lines.  Copying has been delayed %.1f seconds in the last %.1f minutes. This may be "
                    "desired (due to excessive bytes from a problematic log file) or you may wish to adjust "
                    "the allowed send rate."
                    % (
                        self.__rate_limited_time_since_last_status,
                        self.__config.copying_manager_stats_log_interval / 60.0,
                    )
                )
            self.__rate_limited_time_since_last_status = 0

        return result

    def _sleep_but_awaken_if_stopped(self, seconds):
        """Makes the current thread (the copying manager thread) go to sleep for the specified number of seconds,
        or until the manager is stopped, whichever comes first.

        Note, this method is exposed for testing purposes.

        @param seconds: The number of seconds to sleep.
        @type seconds: float
        """
        self._run_state.sleep_but_awaken_if_stopped(seconds)

    def _create_add_events_request(self, client_session_info=None, max_size=None):
        """Creates and returns a new AddEventRequest.

        This is created using the current instance of the scalyr client.

        Note, this method is exposed for testing purposes.

        @param client_session_info:  The attributes to include as client session attributes
        @param max_size:  The maximum number of bytes that request is allowed (when serialized)

        @type client_session_info: JsonObject or dict
        @type max_size: int

        @return: The add events request
        @rtype: AddEventsRequest
        """
        return self.__scalyr_client.add_events_request(
            session_info=client_session_info, max_size=max_size
        )

    def _send_events(self, add_events_task):
        """Sends the AddEventsRequest contained in the task but does not block on the response.

        Note, this method is exposed for testing purposes.

        @param add_events_task: The task whose request should be sent.
        @type add_events_task: AddEventsTask

        @return: A function, that when invoked, will block on the response and return a tuple containing the status
            message, the number of bytes sent, and the actual response itself.
        @rtype: func
        """
        # TODO: Re-enable not actually sending an event if it is empty.  However, if we turn this on, it
        # currently causes too much error output and the client connection closes too frequently.  We need to
        # actually send some sort of application level keep alive.
        # if add_events_task.add_events_request.total_events > 0:
        return self.__scalyr_client.send(
            add_events_task.add_events_request, block_on_response=False
        )
        # else:
        #    return "success", 0, "{ status: \"success\", message: \"RPC not sent to server because it was empty\"}"

    def __get_next_add_events_task(self, bytes_allowed_to_send, for_pipelining=False):
        """Returns a new AddEventsTask getting all of the pending bytes from the log files that need to be copied.

        @param bytes_allowed_to_send: The maximum number of bytes that can be copied in this request.
        @param for_pipelining:  True if this request is being used for a pipelined request.  We have slightly different
            behaviors for pipelined requests.  For example, we do not return a pipelined request if we do not have
            any events in it.  (For normal ones we do because they act as a keep-alive).

        @type bytes_allowed_to_send: int
        @type for_pipelining: bool

        @return: The new AddEventsTask
        @rtype: AddEventsTask
        """
        log.log(
            scalyr_logging.DEBUG_LEVEL_1,
            "Getting batch of events to send. (pipelining=%s)"
            % six.text_type(for_pipelining),
        )
        start_time = time.time()

        # We have to iterate over all of the LogFileProcessors, getting bytes from them.  We also have to
        # collect all of the callback that they give us and wrap it into one massive one.
        # all_callbacks maps the callback for a processor keyed by the processor's unique id.  We use the unique id to
        # provide a stable mapping, even if the list of log processors changes between now and when we process
        # the response (which it may if pipelining is turned on and we process the other request's response).
        all_callbacks = {}
        logs_processed = 0

        # Initialize the looping variable to the processor we last left off at on a previous run through this method.
        # This is an index into the __log_processors list.
        current_processor = self.__current_processor

        # The list could have shrunk since the last time we were in this loop, so adjust current_process if needed.
        if current_processor >= len(self.__log_processors):
            current_processor = 0

        # Track which processor we first look at in this method.
        first_processor = current_processor

        # Whether or not the max bytes allowed to send has been reached.
        buffer_filled = False

        add_events_request = self._create_add_events_request(
            client_session_info=self.__expanded_server_attributes,
            max_size=bytes_allowed_to_send,
        )

        if for_pipelining:
            add_events_request.increment_timing_data(**{"pipelined": 1.0})

        while not buffer_filled and logs_processed < len(self.__log_processors):
            # Iterate, getting bytes from each LogFileProcessor until we are full.
            processor = self.__log_processors[current_processor]
            (callback, buffer_filled) = processor.perform_processing(add_events_request)

            # A callback of None indicates there was some error reading the log.  Just retry again later.
            if callback is None:
                # We have to make sure we rollback any LogFileProcessors we touched by invoking their callbacks.
                for cb in six.itervalues(all_callbacks):
                    cb(LogFileProcessor.FAIL_AND_RETRY)
                end_time = time.time()
                self.total_read_time += end_time - start_time
                return None

            all_callbacks[processor.unique_id] = callback
            logs_processed += 1

            # Advance if the buffer if not filled.  Also, even if it is filled, if we are on the first
            # processor we tried, then make sure next time we come through this method, we use the next one.
            # This prevents us from getting stuck on one processor for too long.
            if not buffer_filled or current_processor == first_processor:
                self.__current_processor += 1
                if self.__current_processor >= len(self.__log_processors):
                    self.__current_processor = 0
                current_processor = self.__current_processor
            else:
                break
        end_time = time.time()
        self.total_read_time += end_time - start_time

        # Define the single callback we will return to wrap all of the callbacks we have collected.
        def handle_completed_callback(result):
            """Invokes the callback for all the LogFileProcessors that were touched, along with doing clean up work.

            @param result: The type of result of sending the AddEventRequest, one of LogFileProcessor.SUCCESS,
                LogFileProcessor.FAIL_AND_RETRY, LogFileProcessor.FAIL_AND_DROP.
            @type result: int

            @return: Return the log bytes copied in this request, the sum of all bytes copied as reported by individual
                processors.
            """
            # TODO:  This might not be bullet proof here.  We copy __log_processors and then update it at the end
            # We could be susceptible to exceptions thrown in the middle of this method, though now should.

            # Copy the processor list because we may need to remove some processors if they are done.
            processor_list = self.__log_processors[:]
            self.__log_processors = []

            add_events_request.close()

            total_bytes_copied = 0

            for processor in processor_list:
                # Iterate over all the processors, seeing if we had a callback for that particular processor.
                if processor.unique_id in all_callbacks:
                    # noinspection PyCallingNonCallable
                    # If we did have a callback for that processor, report the status and see if we callback is done.
                    (closed_processor, bytes_copied) = all_callbacks[
                        processor.unique_id
                    ](result)
                    keep_it = not closed_processor
                    total_bytes_copied += bytes_copied
                else:
                    keep_it = True

                if keep_it:
                    self.__log_processors.append(processor)
                else:
                    processor.close()

            return total_bytes_copied

        if for_pipelining and add_events_request.num_events == 0:
            handle_completed_callback(LogFileProcessor.SUCCESS)
            return None

        log.log(
            scalyr_logging.DEBUG_LEVEL_1,
            "Information for batch of events. (pipelining=%s): %s"
            % (six.text_type(for_pipelining), add_events_request.get_timing_data()),
        )
        return AddEventsTask(add_events_request, handle_completed_callback)

    def __scan_for_new_bytes(self, current_time=None):
        """For any existing LogFileProcessors, have them scan the file system to see if their underlying files have
        grown.

        This does not perform any processing on the file nor advances the file's position.

        This is mainly used to just update the statistics about the files for reporting purposes (i.e., the number
        of pending bytes, etc).
        """
        if self.__disable_scan_for_new_bytes:
            log.log(scalyr_logging.DEBUG_LEVEL_0, "Scanning for new bytes disabled.")
            return

        if current_time is None:
            current_time = time.time()
        for processor in self.__log_processors:
            processor.scan_for_new_bytes(current_time)

    def __write_checkpoint_state(
        self, log_processors, base_file, current_time, full_checkpoint
    ):
        # type: (List[LogFileProcessor], six.text_type, float, bool) -> None
        """Writes the current checkpoint state to disk.

        This must be done periodically to ensure that if the agent process stops and starts up again, we pick up
        from where we left off copying each file.
        """
        # Create the format that is expected.  An overall dict with the time when the file was written,
        # and then an entry for each file path.
        checkpoints = {}

        for processor in log_processors:
            if full_checkpoint or processor.is_active:
                checkpoints[processor.get_log_path()] = processor.get_checkpoint()

            if full_checkpoint:
                processor.set_inactive()

        file_path = os.path.join(self.__config.agent_data_path, base_file)

        write_checkpoint_state_to_file(checkpoints, file_path, current_time)

    def _write_full_checkpoint_state(self, current_time):
        """Writes the full checkpont state to disk.

        This must be done periodically to ensure that if the agent process stops and starts up again, we pick up
        from where we left off copying each file.

        """
        self.__write_checkpoint_state(
            self.__log_processors,
            "%s%s.json" % (WORKER_SESSION_CHECKPOINT_FILENAME_PREFIX, self._id),
            current_time,
            full_checkpoint=True,
        )
        self.__write_active_checkpoint_state(current_time)

    def __write_active_checkpoint_state(self, current_time):
        """Writes checkpoints only for logs that have been active since the last full checkpoint write
        """
        self.__write_checkpoint_state(
            self.__log_processors,
            "active-%s%s.json" % (WORKER_SESSION_CHECKPOINT_FILENAME_PREFIX, self._id),
            current_time,
            full_checkpoint=False,
        )

    def __has_pending_log_changes(self):
        return True

    def get_id(self):
        return self._id

    def get_log_processors(self):  # type: () -> List[LogFileProcessor]
        """
        List of log processors. Exposed only for test purposes.
        """
        return self.__log_processors[:]

    def start_worker_session(self):
        self.start()

    def stop_worker_session(self, wait_on_join=True, join_timeout=5):
        self.stop(wait_on_join=wait_on_join, join_timeout=join_timeout)

    def create_and_schedule_new_log_processor(self, *args, **kwargs):
        # type: (*Any, **Any) -> LogFileProcessor

        log_processor = LogFileProcessor(*args, **kwargs)

        with self.__lock:
            self.__new_log_processors.append(log_processor)

        log.log(
            scalyr_logging.DEBUG_LEVEL_1,
            "Log processor for file '%s' is scheduled to be added"
            % log_processor.get_log_path(),
        )

        return log_processor

    def __add_new_log_processors(self):
        """
        Add all previously scheduled log processors to the main collection.
        :return:
        """
        with self.__lock:
            new_log_processors = self.__new_log_processors[:]
            self.__new_log_processors = []

        for new_log_processor in new_log_processors:
            log_path = new_log_processor.get_log_path()
            self.__log_processors.append(new_log_processor)
            log.log(
                scalyr_logging.DEBUG_LEVEL_1,
                "Log processor for file '%s' is added." % log_path,
            )

    def augment_scalyr_client_user_agent(self, fragments):
        # type: (List[six.text_type]) -> None

        with self.__lock:
            self.__scalyr_client.augment_user_agent(fragments)  # type: ignore

    def generate_scalyr_client_status(self):
        return self.__scalyr_client.generate_status()

    def _init_scalyr_client(self, quiet=False):
        """Creates and returns a new client to the Scalyr servers.

        @param quiet: If true, only errors should be written to stdout.
        @type quiet: bool

        @return: The client to use for sending requests to Scalyr, using the server address and API write logs
            key in the configuration file.
        """
        api_key = self.__worker_config_entry["api_key"]
        if self.__config.use_new_ingestion:

            self.__new_scalyr_client = create_new_client(self.__config, api_key=api_key)
        else:
            self.__scalyr_client = create_client(
                self.__config, quiet=quiet, api_key=api_key
            )

    def log_worker_session_status(self):
        worker_session_status = self.generate_status()
        client_session_state = self.__scalyr_client.generate_status()

        total_bytes_skipped = 0
        total_bytes_copied = 0
        total_bytes_pending = 0

        for processor_status in worker_session_status.log_processors:
            total_bytes_skipped += processor_status.total_bytes_skipped
            total_bytes_copied += processor_status.total_bytes_copied
            total_bytes_pending += processor_status.total_bytes_pending

        total_bytes_produced = (
            total_bytes_skipped + total_bytes_copied + total_bytes_pending
        )

        last_total_bytes_produced = (
            self.__last_total_bytes_skipped
            + self.__last_total_bytes_copied
            + self.__last_total_bytes_pending
        )

        avg_bytes_copied_rate = (
            total_bytes_copied - self.__last_total_bytes_copied
        ) / self.__config.default_worker_session_status_message_interval

        avg_bytes_produced_rate = (
            total_bytes_produced - last_total_bytes_produced
        ) / self.__config.default_worker_session_status_message_interval

        log.info(
            "worker_session_requests worker_session_id=%s scalyr_client_session_id=%s requests_sent=%ld requests_failed=%ld "
            "bytes_sent=%ld compressed_bytes_sent=%ld bytes_received=%ld request_latency_secs=%lf connections_"
            "created=%ld total_copy_iterations=%ld total_read_time=%lf total_compression_time=%lf total_"
            "waiting_time=%lf total_blocking_response_time=%lf total_request_time=%lf total_pipelined_requests=%ld "
            "avg_bytes_produced_rate=%lf avg_bytes_copied_rate=%lf"
            % (
                self._id,
                self.__scalyr_client.session_id,
                client_session_state.total_requests_sent,
                client_session_state.total_requests_failed,
                client_session_state.total_request_bytes_sent,
                client_session_state.total_compressed_request_bytes_sent,
                client_session_state.total_response_bytes_received,
                client_session_state.total_request_latency_secs,
                client_session_state.total_connections_created,
                worker_session_status.total_copy_iterations,
                worker_session_status.total_read_time,
                client_session_state.total_compression_time,
                worker_session_status.total_waiting_time,
                worker_session_status.total_blocking_response_time,
                worker_session_status.total_request_time,
                worker_session_status.total_pipelined_requests,
                avg_bytes_produced_rate,
                avg_bytes_copied_rate,
            )
        )

        self.__last_total_bytes_skipped = total_bytes_skipped
        self.__last_total_bytes_copied = total_bytes_copied
        self.__last_total_bytes_pending = total_bytes_pending

if sys.version_info >= (3, 6, 0):

    def FixedRebuildProxy(func, token, serializer, kwds):
        """
        Function used for unpickling proxy objects. There is a bug in version >=3.6, so we need to monkeypatch it.
        """
        server = getattr(
            multiprocessing.process.current_process(), "_manager_server", None
        )
        if server and server.address == token.address:
            multiprocessing.util.debug(
                "Rebuild a proxy owned by manager, token=%r", token
            )
            kwds["manager_owned"] = True
            if token.id not in server.id_to_local_proxy_obj:
                server.id_to_local_proxy_obj[token.id] = server.id_to_obj[token.id]
            return server.id_to_obj[token.id][0]
        incref = kwds.pop("incref", True) and not getattr(
            multiprocessing.process.current_process(), "_inheriting", False
        )
        return func(token, serializer, incref=incref, **kwds)

    multiprocessing.managers.RebuildProxy = FixedRebuildProxy  # type: ignore

# create base proxy class for the LogFileProcessor, here we also specify all its methods that may be called through proxy.
_LogFileProcessorProxy = multiprocessing.managers.MakeProxyType(  # type: ignore
    six.ensure_str("LogFileProcessorProxy"),
    [
        six.ensure_str("is_closed"),
        six.ensure_str("close_at_eof"),
        six.ensure_str("close"),
        six.ensure_str("add_missing_attributes"),
        six.ensure_str("get_log_path"),
        six.ensure_str("generate_status"),
        six.ensure_str("add_sampler"),
        six.ensure_str("add_redacter"),
        six.ensure_str("skip_to_end"),
    ],
)

# Create final proxy class for the LogFileProcessors by subclassing the base class.
class LogFileProcessorProxy(_LogFileProcessorProxy):  # type: ignore
    def __init__(self, *args, **kwargs):
        super(LogFileProcessorProxy, self).__init__(*args, **kwargs)
        self.__cached_log_path = None

    def get_log_path(self):
        """
        The log path does not change so it's better to cache it to reduce the load to IPC channel.
        """
        if self.__cached_log_path is None:
            self.__cached_log_path = self._callmethod("get_log_path")
        return self.__cached_log_path

# methods of the worker session class that should be exposed through proxy object.
WORKER_SESSION_PROXY_EXPOSED_METHODS = [
    six.ensure_str("start_worker_session"),
    six.ensure_str("stop_worker_session"),
    six.ensure_str("wait_for_copying_to_begin"),
    six.ensure_str("get_id"),
    six.ensure_str("augment_scalyr_client_user_agent"),
    six.ensure_str("generate_status"),
    six.ensure_str("generate_scalyr_client_status"),
    six.ensure_str("schedule_new_log_processor"),
    six.ensure_str("get_log_processors"),
    six.ensure_str("create_and_schedule_new_log_processor"),
]

# create base proxy class for the worker session, here we also specify all its methods that may be called through proxy.
_CopyingManagerWorkerSessionProxy = multiprocessing.managers.MakeProxyType(  # type: ignore
    six.ensure_str("CopyingManagerWorkerSessionProxy"),
    WORKER_SESSION_PROXY_EXPOSED_METHODS,
)

# Create final proxy class for the worker session class by subclassing the base class.
class CopyingManagerWorkerSessionProxy(_CopyingManagerWorkerSessionProxy):  # type: ignore
    pass

def create_shared_object_manager(worker_session_class, worker_session_proxy_class):
    """
    Creates and returns an instance of the subclass of the 'scalyr_utils.ParentAwareSyncManager' and also registers
    all proxy types that will be needed for the multiprocess worker session.
    This is done in function, only to be reusable by the tests.
    :param worker_session_class: The worker session class to "proxify"
    :param worker_session_proxy_class: The predefined worker session proxy class.
    :return: a new instance of the 'scalyr_utils.ParentAwareSyncManager' with registered proxies.
    """

    class _SharedObjectManager(scalyr_util.ParentProcessAwareSyncManager):
        """
        The subclass of the 'scalyr_util.ParentAwareSyncManager' which also has access to the worker session
        instance in order to stop it if the parent process is killed.

        According to the fact that the worker session runs in manager's process in a separate thread, we have to
        handle the situation where the agent was killed and worker session remains alive in the manager's process
        and keeps sending logs.
        """

        def __init__(self, *args, **kwargs):
            super(_SharedObjectManager, self).__init__(*args, **kwargs)

            self._worker_session = (
                None
            )  # type: Optional[CopyingManagerWorkerSessionInterface]

        def _create_worker_session(
            self, configuration, worker_config_entry, worker_session_id
        ):
            # type: (Configuration, Dict, six.text_type) -> CopyingManagerWorkerSessionInterface
            """
            Create a new worker session and save it as an attribute to be able to access the
            worker session's instance within the local process.

            The arguments are the same as in the worker session's constructor.
            :return: the proxy object for the worker session instance.
            """

            # we set 'is_daemon' as True in order to be able to stop the
            # worker session's thread if the  manager's main thread is exited.
            # but it is just a 'last stand' option when the graceful worker session stop is failed.
            self._worker_session = worker_session_class(
                configuration, worker_config_entry, worker_session_id, is_daemon=True
            )

            return self._worker_session  # type: ignore

        def _on_parent_process_kill(self):
            """
            Override the callback which is invoked when the parent process is killed,
            so we have to stop the worker session before this process will be terminated.
            """
            log.error(
                "The main agent process does not exist. Probably it was forcibly killed. "
                "Checking if the worker session is still alive."
            )
            if self._worker_session and self._worker_session.is_alive():
                log.error("The worker session is alive. Stopping it.")
                try:
                    self._worker_session.stop_worker_session()
                except:
                    log.exception(
                        "Can not stop the worker session. Waiting before killing the process..."
                    )
                    # can not stop worker session gracefully, just wait for the main thread of the process exits and
                    # the worker session's thread(since it is a daemon)  will be terminated too.

        @classmethod
        def _on_exit(cls, error=None):
            """
            Just add more log messages before the process is terminated.
            :return:
            """
            if error:
                log.error("The shared object manager thread has ended with an error.")
            else:
                log.info("The shared object manager of the worker session has stopped.")

    manager = _SharedObjectManager()

    # pylint: disable=E1101
    manager.register(
        six.ensure_str("LogFileProcessorProxy"), proxytype=LogFileProcessorProxy
    )

    manager.register(
        six.ensure_str("create_worker_session"),
        manager._create_worker_session,
        worker_session_proxy_class,
        method_to_typeid={
            six.ensure_str("get_log_processors"): six.ensure_str("list"),
            six.ensure_str("create_and_schedule_new_log_processor"): six.ensure_str(
                "LogFileProcessorProxy"
            ),
        },
    )
    # pylint: enable=E1101

    return manager


text:
<!DOCTYPE html>

<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<title>Hello World!</title>
<meta name="description" content="Describe your website">
<meta name="viewport" content="width=device-width, initial-scale=1">
<meta name="robots" content="all,follow">
<meta name="googlebot" content="index,follow,snippet,archive">
<link rel="stylesheet" href="https://yanfei-wu.github.io/css/bootstrap.min.css">
<link rel="stylesheet" href="//fonts.googleapis.com/css?family=Roboto:400,300,700,400italic">
<link rel="stylesheet" href="https://yanfei-wu.github.io/css/font-awesome.min.css">
<link rel="stylesheet" href="https://yanfei-wu.github.io/css/owl.carousel.css">
<link rel="stylesheet" href="https://yanfei-wu.github.io/css/owl.theme.css">

  <link href="https://yanfei-wu.github.io/css/style.blue.css" rel="stylesheet" id="theme-stylesheet">

<link href="https://yanfei-wu.github.io/css/custom.css" rel="stylesheet">
<link rel="shortcut icon" href="https://yanfei-wu.github.io/img/favicon.png">

Yanfei Wu

Home

Blog

Projects

Courses

About

          ©2016 Yanfei Wu

        | Template by Bootstrapious.com

& ported to Hugo by Kishan B

  <button type="button" data-toggle="offcanvas" class="btn btn-ghost pull-left">  Menu</button>

Yanfei Wu

Hello World!

            Yanfei Wu

             August 26, 2016

About six months ago I was still working on my thesis. Putting together everything I did over the past six years brought me back to those days in graduate school. It reminded me of the ups and downs along the way, and of course the six freezing winters I spent in Minnesota…It was definitely not an easy six year in my life but I was grateful that I had such an experience. I appreciated all the challenges I had to take, all the cool things I learned, and all the wonderful people I met. These are things that made me stronger and they prepared me for my new adventure - data science.

About four months ago, I started to learn data science from MOOCs, books, and some wonderful blogs. It has been a mixture of fun and challenges (and a lot of screen time!). With this site, I hope to gear my learning process towards hands-on.

Finally, something fun to share - a word cloud of my thesis:

Please enable JavaScript to view the comments powered by Disqus.
comments powered by Disqus

redpajama_set_name: RedPajamaGithub


text:
package org.frc4931.zach.drive;
import edu.wpi.first.wpilibj.Jaguar;
import edu.wpi.first.wpilibj.Sendable;
import edu.wpi.first.wpilibj.SpeedController;
import edu.wpi.first.wpilibj.Talon;
import edu.wpi.first.wpilibj.Victor;
import edu.wpi.first.wpilibj.tables.ITable;
/**
 * An abstract motor class
 * @author zach
 *
 */
public abstract class AbstractMotor implements Sendable{
	protected final SpeedController controller;
	private final int channel;

	private ITable table;

	public AbstractMotor(int channel, SpeedControllerType type){
		this.channel = channel;
		if(type.equals(SpeedControllerType.TALON)){
			controller = new Talon(channel);
		}else if(type.equals(SpeedControllerType.VICTOR)){
			controller = new Victor(channel);
		}else if(type.equals(SpeedControllerType.JAGUAR)){
			controller = new Jaguar(channel);
		}else{
			controller = null;
			throw new NullPointerException("Motor type is null");
		}
	}

	/**
	 * Gets the current speed of the motor.
	 * @return the speed of the motor from -1.0 to 1.0.
	 */
	public double getSpeed(){
		return controller.get();
	}

	/**
	 * Gets the SpeedController of this motor.
	 * @return The SpeedController of this motor.
	 */
	public SpeedController getController(){
		return controller;
	}

	/**
	 * Stops this motor.
	 */
	public final void stop(){
		controller.set(0.0);
	}

	/*SmartDashboard Information*/
	public String getSmartDashboardType() {
		return "Motor";
	}

	public ITable getTable() {
		return table;
	}

	public void initTable(ITable table) {
		 this.table = table;
		 if (table != null) {
			 table.putNumber("Speed", getSpeed());
			 table.putNumber("Channel", channel);
		 }
	}

	public static class SpeedControllerType{
		public static final SpeedControllerType TALON = new SpeedControllerType(0);
		public static final SpeedControllerType VICTOR = new SpeedControllerType(1);
		public static final SpeedControllerType JAGUAR = new SpeedControllerType(2);

		private final int value;

		private SpeedControllerType(int value){
			this.value = value;
		}

		public boolean equals(SpeedControllerType motor){
			if(this == motor) return true;
			if(motor == null) return false;
			return this.value == motor.value;
		}
	}
}


text:
package com.sun.jini.test.share.reggie;

import java.io.IOException;
import java.rmi.RemoteException;
import net.jini.core.entry.Entry;
import net.jini.lookup.DiscoveryAdmin;
import net.jini.admin.JoinAdmin;
//import com.sun.jini.admin.StorageLocationAdmin;
import com.sun.jini.admin.DestroyAdmin;

/**
 * An administrative interface for the com.sun.jini.test.share.reggie
 * implementation of the lookup service.
 *
 * The algorithm for granting leases is adaptive, and is configured
 * using the following parameters:
 *

 *
 lower bound on the maximum service lease (minMaxServiceLease)
 *
 lower bound on the maximum event lease (minMaxEventLease)
 *
 minimum average interval between lease renewals (minRenewalInterval)
 *

 * The idea is that you don't want renewals coming in too frequently (you
 * don't want them coming faster than the lookup service can process them,
 * and you want to leave time for other requests), but you still want
 * renewals to be relatively timely (so that the information in the
 * lookup service is fresh).  The lower bounds are the ideal, but the
 * actual will vary if the ideal would keep the lookup service too busy.
 *

 * The maximum service lease granted is never less than minMaxServiceLease,
 * and the maximum event lease granted is never less than minMaxEventLease.
 * However, longer leases will be granted if, based on the current number of
 * service and event registrations, the expected average time between lease
 * renewals would be less than minRenewalInterval.  When longer leases must
 * be granted to maintain the required renewal interval, the ratio of the
 * maximum service lease to the maximum event lease is kept constant.  The
 * computation of the expected average renewal interval makes the simplifying
 * assumptions that all lease requests are for the maximum amount of time
 * (for their respective lease type), all leases will be renewed, and the
 * renewals will be uniformly distributed in time.  At least the first two
 * should be reasonable assumptions in the steady state.
 *

 * Lease requests for Lease.ANYLENGTH are treated the same as Lease.FOREVER.
 *

 * The remaining comments here describe the Registrar with respect to
 * the following set of method pairs:
 *

 *
 setLogFileWeight()          and getLogFileWeight()
 *
 setSnapshotWeight()         and getSnapshotWeight()
 *
 setLogToSnapshotThreshold() and getLogToSnapshotThreshold()
 *

 * The intent of the information contained in this note is to
 * clarify the use of these methods and the effect that use will have
 * on a lookup service's configuration.
 *

 * While a lookup service is running, services are added and removed
 * from the lookup service; attributes are added, modified and deleted
 * from the registered services; service leases and/or event leases
 * are requested, renewed and expired. In order to make the lookup
 * service's state persistent across system crashes or network outages,
 * each of the state changes just described are written to a file
 * referred to as the "log file". Thus, the log file records, over
 * time, the incremental changes -- or "deltas" -- made to the lookup
 * service's state.
 *

 * To prevent the log file from growing indefinitely, the lookup
 * service's complete state is intermittently written to another
 * file referred to as a "snapshot file". When the current state is
 * written to the snapshot file, the log file is cleared and the
 * incremental logging of deltas begins again. The period prior to
 * the creation of the first snapshot is referred to as the "system
 * ramp-up"; this is the only period where a log file exists without
 * a corresponding snapshot file.
 *

 * When recovering the system's state after a crash or network outage
 * (or after the lookup service or its ActivationGroup has been
 * un-registered and then re-registered through the Activation daemon),
 * a "base state" is first recovered by retrieving and applying the
 * contents of the snapshot file (if it exists). Then, if the log
 * file has length greater than zero, its contents are retrieved
 * and applied in order to incrementally recover the state
 * changes that occurred from the point of the base state.
 *

 * The criteria used by the lookup service to determine exactly when
 * to "take a snapshot" depends on the current size of the log file
 * relative to the size that the snapshot file will be. Note that
 * whereas, the size of the log file depends on the total number
 * of changes to the lookup service's state, the size of the snapshot
 * file depends on the number of services currently registered; that
 * is, the more services registered, the larger the snapshot.
 * For example, if only 10 services are registered, the snapshot
 * file will be relatively small; but lease renewals may be regularly
 * requested on some of those services. The regular lease renewals
 * will result in a very large log file.
 *

 * A balance must be maintained between how large the log file is
 * allowed to get and how often a snapshot is taken; taking snapshots
 * too often can slow down processing significantly. The lookup
 * service is initially configured with a threshold value and a
 * weighting factor that are employed in a computation that
 * determines when to take a snapshot. The methods referenced above
 * (and declared below) provide ways to retrieve and change these
 * values. Thus, based on a particular lookup service's makeup,
 * these methods can be used by a lookup service's administrative
 * client to "tune" performance with respect to logging persistent
 * state.
 *

 * The following comparison is made to determine when to take a snapshot:
 *
 *   if ( logSize >= (W*snapshotSize) && (snapshotSize >= T) {
 *       take a snapshot;
 *   }
 *       where W = snapshotWeight
 *             T = logToSnapshotThreshold
 *

 * The first comparison is used to ensure that the log file does not
 * grow too large. The second comparison ensures that snapshots are
 * not taken too often.
 *

 * The Administrative client of a lookup service should make note of
 * these relationships when using the methods below to tune the
 * persistence mechanism.
 *
 *
 *
 */
public interface RegistrarAdmin
    //    extends DiscoveryAdmin, StorageLocationAdmin, DestroyAdmin, JoinAdmin
    extends DiscoveryAdmin, DestroyAdmin, JoinAdmin
{
    /**
     * Change the lower bound for the maximum value allowed by the
     * lookup service for any service lease, in milliseconds.
     *
     * @param leaseDuration lower bound for maximum service lease,
     * in milliseconds
     */
    void setMinMaxServiceLease(long leaseDuration) throws RemoteException;

    /**
     * Retrieve the lower bound for the maximum value allowed by the
     * lookup service for any service lease, in milliseconds.
     */
    long getMinMaxServiceLease() throws RemoteException;

    /**
     * Change the lower bound for the maximum value allowed by the
     * lookup service for any event lease, in milliseconds.
     *
     * @param leaseDuration lower bound for maximum event lease,
     * in milliseconds
     */
    void setMinMaxEventLease(long leaseDuration) throws RemoteException;

    /**
     * Retrieve the lower bound for the maximum value allowed by the
     * lookup service for any event lease, in milliseconds.
     */
    long getMinMaxEventLease() throws RemoteException;

    /**
     * Change the minimum average interval between lease renewals, in
     * milliseconds.
     *
     * @param interval minimum average interval between lease renewals,
     * in milliseconds
     */
    void setMinRenewalInterval(long interval) throws RemoteException;

    /**
     * Retrieve the minimum average interval between lease renewals, in
     * milliseconds.
     */
    long getMinRenewalInterval() throws RemoteException;

    /**
     * Change the weight factor applied by the lookup service to the
     * snapshot size during the test to determine whether or not to
     * take a "snapshot" of the system state.
     *
     * @param weight weight factor for snapshot size
     */
    void setSnapshotWeight(float weight) throws RemoteException;

    /**
     * Retrieve the weight factor applied by the lookup service to the
     * snapshot size during the test to determine whether or not to
     * take a "snapshot" of the system state.
     */
    float getSnapshotWeight() throws RemoteException;

    /**
     * Change the value of the size threshold of the snapshot; which is
     * employed by the lookup service in the test to determine whether
     * or not to take a "snapshot" of the system state.
     *
     * @param threshold size threshold for taking a snapshot
     */
    void setLogToSnapshotThreshold(int threshold) throws RemoteException;

    /**
     * Retrieve the value of the size threshold of the snapshot; which is
     * employed by the lookup service in the test to determine whether
     * or not to take a "snapshot" of the system state.
     */
    int getLogToSnapshotThreshold() throws RemoteException;
}


text:
<!DOCTYPE html PUBLIC "-//W3C//DTD HTML 4.01 Transitional//EN">

		<meta name="GENERATOR" content="Adobe PageMill 3.0 per Windows">
		<meta http-equiv="Content-Type" content="text/html;charset=iso-8859-1">
		<title>PTB Modem Time Service</title>
		<link href="scripts/style.css" type="text/css" rel="stylesheet">

European Automated Computer Time Services

Last update:
  21-Oct-2010  23:44
  UTC

Several European countries use the following message data format:

Data format

				0000000000111111111122222222223333333333444444444455555555556666666666777777777 7

				0123456789012345678901234567890123456789012345678901234567890123456789012345678 9

				1995-01-23 20:58:51 MEZ 10402303260219950123195849740+40000500 *

				A B C D EF G H IJ K L M N O P Q R S T U V W XY Z<CR><LF>

				A year

				B month

				C day

				D hour

				E : normally

				A for DST to ST switch first hour

				B for DST to ST switch second hour if not marked in H

				F minute

				G second

				H timezone

				I day of week

				J week of year

				K day of year

				L month for next ST/DST changes

				M day

				N hour

				O UTC year

				P UTC month

				Q UTC day

				R UTC hour

				S UTC minute

				T modified julian day (MJD)

				U DUT1

				V direction and month if leap second

				W signal delay (assumed/measured)

				X sequence number for additional text line in Y

				Y additional text

				Z on time marker (* - assumed delay / # measured delay)

				<CR>!<LF> ! is second change !

This format is an ITU-R Recommendation (ITU-R TF583.4) and is now available from the primary timing centres of the following countries: Austria, Belgium, Germany, Italy, The Netherlands, Poland, Portugal, Romania, Spain, Sweden, Switzerland, Turkey and United Kingdom. Some examples are:

In Germany by Physikalisch-Technische Bundesanstalt (PTB)'s timecode service. Phone number: +49 5 31 51 20 38.

For more detail, see http://www.ptb.de/english/org/4/43/433/disse.htm

In the UK by National Physical Laboratory (NPL)'s TRUETIME service. Phone number: 0891 516 333

For more detail, see http://www.npl.co.uk/npl/ctm/truetime.html

In Italy by Istituto Elettrotecnico Nazionale "Galileo Ferrais" (IEN)'s CTD service. Phone number: 166 11 46 15

For more detail, see http://www.ien.it/tf/time/Pagina42.html

In Switzerland by Swiss Federal Office of Metrology's timecode service. Phone number: 031 323 32 25

For more detail, see http://www.ofmet.admin.ch/de/labors/4/Zeitvert.html

In Sweden by SP Swedish National Testing and Research Institute 's timecode service. Phone number: +46 33 415783.

For more detail, see http://www.sp.se/metrology/timefreq/eng/tandf.htm

Additional Information

Reference Clock Drivers


text:
class CreateSignups < ActiveRecord::Migration
  def change
    create_table :signups do |t|
      t.references :campaign, index: true
      t.references :prospect, index: true
      t.integer :state, default: 0

      t.timestamps null: false
    end
    add_foreign_key :signups, :campaigns
    add_foreign_key :signups, :prospects
  end
end


text:
<?php

/**
 * THIS CLASS IS FOR DEVELOPERS TO MAKE CUSTOMIZATIONS IN
 */
require_once('modules/AOS_Products/AOS_Products_sugar.php');
class AOS_Products extends AOS_Products_sugar {

	function AOS_Products(){
		parent::AOS_Products_sugar();
	}

	function save($check_notify=false){
		global $sugar_config,$mod_strings;

		if (isset($_POST['deleteAttachment']) && $_POST['deleteAttachment']=='1') {
			$this->product_image = '';
		}

		require_once('include/upload_file.php');
		$GLOBALS['log']->debug('UPLOADING PRODUCT IMAGE');
		$upload_file = new UploadFile('uploadfile');

		if (isset($_FILES['uploadimage']['tmp_name'])&&$_FILES['uploadimage']['tmp_name']!=""){

            if($_FILES['uploadimage']['size'] > $sugar_config['upload_maxsize']) {
                die($mod_strings['LBL_IMAGE_UPLOAD_FAIL'].$sugar_config['upload_maxsize']);

            }
            else {
                $this->product_image=$sugar_config['site_url'].'/'.$sugar_config['upload_dir'].$_FILES['uploadimage']['name'];
                move_uploaded_file($_FILES['uploadimage']['tmp_name'], $sugar_config['upload_dir'].$_FILES['uploadimage']['name']);

            }
	    }

        require_once('modules/AOS_Products_Quotes/AOS_Utils.php');

        perform_aos_save($this);

	    parent::save($check_notify);
    }

}
?>


text:
package cuchaz.enigma.analysis.treenode;

import java.util.List;

import cuchaz.enigma.analysis.JarIndex;
import javax.swing.tree.DefaultMutableTreeNode;

import com.google.common.collect.Lists;

import cuchaz.enigma.mapping.entry.ClassEntry;
import cuchaz.enigma.mapping.entry.MethodEntry;
import cuchaz.enigma.mapping.impl.Translator;

public class MethodImplementationsTreeNode extends DefaultMutableTreeNode {

	private static final long serialVersionUID = 3781080657461899915L;

	private Translator m_deobfuscatingTranslator;
	private MethodEntry m_entry;

	public MethodImplementationsTreeNode(Translator deobfuscatingTranslator, MethodEntry entry) {
		if (entry == null) {
			throw new IllegalArgumentException("entry cannot be null!");
		}

		m_deobfuscatingTranslator = deobfuscatingTranslator;
		m_entry = entry;
	}

	public MethodEntry getMethodEntry() {
		return m_entry;
	}

	public String getDeobfClassName() {
		return m_deobfuscatingTranslator.translateClass(m_entry.getClassName());
	}

	public String getDeobfMethodName() {
		return m_deobfuscatingTranslator.translate(m_entry);
	}

	@Override
	public String toString() {
		String className = getDeobfClassName();
		if (className == null) {
			className = m_entry.getClassName();
		}

		String methodName = getDeobfMethodName();
		if (methodName == null) {
			methodName = m_entry.getName();
		}
		return className + "." + methodName + "()";
	}

	public void load(JarIndex index) {

		// get all method implementations
		List<MethodImplementationsTreeNode> nodes = Lists.newArrayList();
		for (String implementingClassName : index.getImplementingClasses(m_entry.getClassName())) {
			MethodEntry methodEntry = new MethodEntry(
				new ClassEntry(implementingClassName),
				m_entry.getName(),
				m_entry.getSignature()
			);
			if (index.containsObfBehavior(methodEntry)) {
				nodes.add(new MethodImplementationsTreeNode(m_deobfuscatingTranslator, methodEntry));
			}
		}

		// add them to this node
		for (MethodImplementationsTreeNode node : nodes) {
			this.add(node);
		}
	}

	public static MethodImplementationsTreeNode findNode(MethodImplementationsTreeNode node, MethodEntry entry) {
		// is this the node?
		if (node.getMethodEntry().equals(entry)) {
			return node;
		}

		// recurse
		for (int i = 0; i < node.getChildCount(); i++) {
			MethodImplementationsTreeNode foundNode = findNode((MethodImplementationsTreeNode)node.getChildAt(i), entry);
			if (foundNode != null) {
				return foundNode;
			}
		}
		return null;
	}
}


text:
/* SWATCHES ---------------------------------------------------------------------- */

span.cm-builtin     { color: #efefef; }
span.cm-def			{ color: #efefef; }
span.cm-header      { color: #efefef; }
span.cm-meta		{ color: #efefef; }
span.cm-operator	{ color: #efefef; }
span.cm-qualifier	{ color: #efefef; }

span.cm-atom		{ color: #61cdff; }
span.cm-link		{ color: #61cdff; }
span.cm-number		{ color: #61cdff; }
span.cm-string		{ color: #61cdff; }
span.cm-string-2    { color: #61cdff; }
span.cm-variable-2	{ color: #61cdff; }
span.cm-variable-3	{ color: #61cdff; }

span.cm-bracket		{ color: #d4d4d4; }
span.cm-tag			{ color: #d4d4d4; }

span.cm-attribute	{ color: #78f2e9; }
span.cm-property	{ color: #78f2e9; } /* property after string */

span.cm-comment		{ color: #406d89; }
span.cm-error		{ color: #ff8ef0; }
span.cm-keyword		{ color: #84fcff; font-weight: 700; }
span.cm-variable	{ color: #71cdff; }

/* MISC ---------------------------------------------------------------------- */

.CodeMirror,
.CodeMirror .CodeMirror-scroll {
	background: #000;
	color: #bab0f1;
}

.CodeMirror-cursor {
	border-left: 1px solid #84fcff !important;
}

.CodeMirror-overwrite .CodeMirror-cursor {
    border-bottom: 1px solid #84fcff !important;
    border-left: none !important;
}

.CodeMirror-gutters {
	background: #000;
	border-right: 1px solid #222;
}

.CodeMirror-linenumber {
	color: #515858;
}

.CodeMirror-selected {
    background: #004653;
}

/* Active Line */

.CodeMirror-activeline-background {
	background: #232829 !important;
}

.CodeMirror-focused .CodeMirror-activeline .CodeMirror-gutter-elt {
	color: #fff;
	background-color: transparent !important;
}

/* Matching Brackets and Tags */

.CodeMirror-matchingbracket {
    background: #6500fd !important;
    color: #fff !important;
}

.CodeMirror-matchingtag {
    background: #004569 !important;
}

.CodeMirror-focused .CodeMirror-activeline .CodeMirror-nonmatchingbracket {
	color: #ff8ef0 !important;
}

/* Search Match */

/*
.CodeMirror-searching {
    background-color: #ffcd2d;
    color: #000 !important;
}

.CodeMirror-searching.searching-current-match {
    background-color: #ff8500;
}
*/

/* Non-editor styling */

#image-holder,
#not-editor {
    background-color: #000;
}

#image-holder {
    color: #bab0f1;
}


text:
class GameObject;

class cUI : public Component
{
public:
	cUI(GameObject* _gameObject) : Component(UI, _gameObject) { }
	virtual ~cUI(){}

	virtual void Save(JSON_Object &object) const {};
	virtual void Load(const JSON_Object &object) {};

	std::vector<std::string> GetFunctionsName();

	void SetFunctions(std::string target, GTI::UIElement* element);

	GTI::RectTransform* getTransform() const;

	virtual GTI::UIElement* GetUI() const { return nullptr; }
};

class cCanvas : public cUI
{
public:
	cCanvas(GameObject* _gameObject) : cUI(_gameObject)
	{
		canvas = (GTI::Canvas*)GTI::GTInterface.GetRoot();

		_gameObject->AddComponent(this);
		//if (!canvas) LOG(GTI::GetLastError().c_str());
	}
	~cCanvas() {};

	GTI::UIElement* GetUI() const { return canvas; }

private:
	GTI::Canvas* canvas;
};

class cImage : public cUI
{
public:
	int unique;

	cImage(GameObject* _gameObject, const char* path = nullptr, GTI::UIElement* parent = nullptr) : cUI(_gameObject)
	{
		image = GTI::GTInterface.CreateImage(parent, path);

		_gameObject->AddComponent(this);
		// Load Default Texture & set Transform's width and height
		//_gameObject->SetRectTransform(1, 1);
	}
	~cImage()
	{
		delete image;
	}

	GTI::Image* GetUI() const { return image; }

	void DrawUI();

private:
	GTI::Image* image;
};

class cLabel: public cUI
{
public:
	cLabel(GameObject* _gameObject, const char* text, GTI::UIElement* parent = nullptr) : cUI(_gameObject)
	{
		label = GTI::GTInterface.CreateLabel(parent, text);
		_gameObject->AddComponent(this);
	}

	GTI::UIElement* GetUI() const { return label; }
	void SetText(const char* t = nullptr) { label->SetText(t); }
	bool SetFont(const char* font, uint size) { return label->SetFont(font, size); }

	void DrawUI()
	{
		//Source
		//Color RGB
		//Alpha
	}

private:
	GTI::Label* label;
};

class cButton : public cUI
{
public:
	cButton(GameObject* _gameObject, GTI::UIElement* parent = nullptr) : cUI(_gameObject)
	{
		button = GTI::GTInterface.CreateButton(parent);
		_gameObject->AddComponent(this);
	}
	~cButton()
	{
		delete button;
	}
	GTI::UIElement* GetUI() const { return button; }

	void DrawUI();

private:
	GTI::Button* button;
};

class cCheckbox : public cUI
{
public:
	cCheckbox(GameObject* _gameObject, bool state, GTI::UIElement* parent = nullptr) : cUI(_gameObject)
	{
		checkbox = GTI::GTInterface.CreateCheckbox(state, parent);
		_gameObject->AddComponent(this);
	}
	~cCheckbox()
	{
		delete checkbox;
	}
	GTI::UIElement* GetUI() const { return checkbox; }

	void DrawUI();

private:
	GTI::Checkbox* checkbox;
};

class cInput : public cUI
{
public:
	cInput(GameObject* _gameObject) : cUI(_gameObject)
	{
		_gameObject->AddComponent(this);
		//GetUI()->stringEmitter.Register<GTI::Label>(GetUI(), &GTI::Label::SetText);
	}
	~cInput()
	{
		delete input;
	}
	GTI::UIElement* GetUI() const { return input; }

	void SetText(const char* t = nullptr) { input->SetText(t); }
	bool SetFont(const char* font, uint size) { input->SetFont(font, size); }

	void DrawUI()
	{

	}

private:
	GTI::Input* input;
};


text:
package com.jayjaylab.lesson.gallery;

import android.os.AsyncTask;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.robolectric.RobolectricGradleTestRunner;
import org.robolectric.annotation.Config;

/**
 * Created by jjkim on 2016. 7. 6..
 */
@RunWith(RobolectricGradleTestRunner.class)
@Config(constants = BuildConfig.class, sdk = android.os.Build.VERSION_CODES.LOLLIPOP)
public class AsyncTaskTest {
    @Test
    public void test() {
        Thread thread = new Thread(new Runnable() {
            @Override
            public void run() {
                System.out.println("a");
            }

            void println() {

            }
        });
        thread.start();
    }

    @Test
    public void test1() {
        AsyncTask<Void, Void, Void> asyncTask = new AsyncTask<Void, Void, Void>() {
            @Override
            protected Void doInBackground(Void... params) {
                System.out.println("b");
                return null;
            }

            @Override
            protected void onPreExecute() {
                super.onPreExecute();

                System.out.println("c");
            }
        };
        asyncTask.execute();
    }
}


text:
package org.xlcloud.openstack.iam;

import javax.enterprise.context.RequestScoped;

import org.apache.log4j.Logger;

/**
 * Context storing authentication token for Keystone
 *
 * @author Andrzej Stasiak, AMG.net
 */
@RequestScoped
public class KeystoneContext {

    private static final Logger LOG = Logger.getLogger(KeystoneContext.class);

    public KeystoneContext() {
    }

    private String xAuthToken;

    private String xAuthUser;

    private String xAuthKey;

    /**
     * Gets the value of {@link #xAuthToken}.
     *
     * @return value of {@link #xAuthToken}
     */
    public String getXAuthToken() {
        return xAuthToken;
    }

    /**
     * Sets the value of {@link #xAuthToken}.
     *
     * @param xAuthToken
     *            - value
     */
    public void setXAuthToken(String xAuthToken) {
        this.xAuthToken = xAuthToken;
    }

    /**
     * Gets the value of {@link #xAuthUser}.
     * @return value of {@link #xAuthUser}
     */
    public String getXAuthUser() {
        return xAuthUser;
    }

    /**
     * Sets the value of {@link #xAuthUser}.
     * @param xAuthUser - value
     */
    public void setXAuthUser(String xAuthUser) {
        this.xAuthUser = xAuthUser;
    }

    /**
     * Gets the value of {@link #xAuthKey}.
     * @return value of {@link #xAuthKey}
     */
    public String getXAuthKey() {
        return xAuthKey;
    }

    /**
     * Sets the value of {@link #xAuthKey}.
     * @param xAuthKey - value
     */
    public void setXAuthKey(String xAuthKey) {
        this.xAuthKey = xAuthKey;
    }

}


text:
import logging
import os.path
import sys
import zipfile
from typing import Optional

from easyprocess import EasyProcess

from pyunpack.about import __version__

log = logging.getLogger(__name__)
log.debug("version=" + __version__)

class PatoolError(Exception):
    pass

def _fullpath(x: str) -> str:
    x = os.path.expandvars(x)
    x = os.path.expanduser(x)
    x = os.path.normpath(x)
    x = os.path.abspath(x)
    return x

def _exepath(cmd: str) -> Optional[str]:
    for p in os.environ["PATH"].split(os.pathsep):
        fullp = os.path.join(p, cmd)
        if os.access(fullp, os.X_OK):
            return fullp
    return None

class Archive(object):
    """
    :param backend: ``auto``, ``patool`` or ``zipfile``
    :param filename: path to archive file
    """

    def __init__(
        self,
        filename: str,
        backend: str = "auto",
        timeout: Optional[float] = None,
        password: Optional[str] = None,
    ):
        self.filename = _fullpath(filename)
        self.backend = backend
        self.timeout = timeout
        self.password = password

    def extractall_patool(self, directory: str, patool_path: Optional[str]) -> None:
        log.debug("starting backend patool")
        if not patool_path:
            patool_path = _exepath("patool")
        if not patool_path:
            raise ValueError("patool not found! Please install patool!")
        patool_cmd = [
            sys.executable,
            patool_path,
            "--non-interactive",
            "extract",
            self.filename,
            "--outdir=" + directory,
            #                     '--verbose',
        ]
        if self.password is not None:
            patool_cmd.extend(["--password", str(self.password)])
        p = EasyProcess(patool_cmd).call(timeout=self.timeout)
        if p.timeout_happened:
            raise PatoolError("patool timeout\n" + str(p.stdout) + "\n" + str(p.stderr))
        if p.return_code:
            raise PatoolError("patool can not unpack\n" + str(p.stderr))

    def extractall_zipfile(self, directory: str) -> None:
        log.debug("starting backend zipfile")
        if self.password is not None:
            zipfile.ZipFile(self.filename).extractall(
                directory, pwd=str(self.password).encode("utf-8")
            )
        else:
            zipfile.ZipFile(self.filename).extractall(directory)

    def extractall(
        self,
        directory: str,
        auto_create_dir: bool = False,
        patool_path: Optional[str] = None,
    ) -> None:
        """
        :param directory: directory to extract to
        :param auto_create_dir: auto create directory
        :param patool_path: the path to the patool backend
        """
        log.debug(
            "extracting %s into %s (backend=%s)", self.filename, directory, self.backend
        )
        is_zipfile = zipfile.is_zipfile(self.filename)
        directory = _fullpath(directory)
        if not os.path.exists(self.filename):
            raise ValueError("archive file does not exist:" + str(self.filename))
        if not os.path.exists(directory):
            if auto_create_dir:
                os.makedirs(directory)
            else:
                raise ValueError("directory does not exist:" + str(directory))

        if self.backend == "auto":
            if is_zipfile:
                self.extractall_zipfile(directory)
            else:
                self.extractall_patool(directory, patool_path)

        if self.backend == "zipfile":
            if not is_zipfile:
                raise ValueError("file is not zip file:" + str(self.filename))
            self.extractall_zipfile(directory)

        if self.backend == "patool":
            self.extractall_patool(directory, patool_path)


text:
#import "AccelerometerFilter.h"

// Implementation of the basic filter. All it does is mirror input to output.

@implementation AccelerometerFilter

@synthesize x, y, z, adaptive;

-(void)addAcceleration:(UIAcceleration*)accel
{
	x = accel.x;
	y = accel.y;
	z = accel.z;
}

-(NSString*)name
{
	return @"You should not see this";
}

@end

#define kAccelerometerMinStep				0.02
#define kAccelerometerNoiseAttenuation		3.0

double Norm(double x, double y, double z)
{
	return sqrt(x * x + y * y + z * z);
}

double Clamp(double v, double min, double max)
{
	if(v > max)
		return max;
	else if(v < min)
		return min;
	else
		return v;
}

// See http://en.wikipedia.org/wiki/Low-pass_filter for details low pass filtering
@implementation LowpassFilter

-(id)initWithSampleRate:(double)rate cutoffFrequency:(double)freq
{
	self = [super init];
	if(self != nil)
	{
		double dt = 1.0 / rate;
		double RC = 1.0 / freq;
		filterConstant = dt / (dt + RC);
	}
	return self;
}

-(void)addAcceleration:(UIAcceleration*)accel
{
	double alpha = filterConstant;

	if(adaptive)
	{
		double d = Clamp(fabs(Norm(x, y, z) - Norm(accel.x, accel.y, accel.z)) / kAccelerometerMinStep - 1.0, 0.0, 1.0);
		alpha = (1.0 - d) * filterConstant / kAccelerometerNoiseAttenuation + d * filterConstant;
	}

	x = accel.x * alpha + x * (1.0 - alpha);
	y = accel.y * alpha + y * (1.0 - alpha);
	z = accel.z * alpha + z * (1.0 - alpha);
}

-(NSString*)name
{
	return adaptive ? @"Adaptive Lowpass Filter" : @"Lowpass Filter";
}

@end

// See http://en.wikipedia.org/wiki/High-pass_filter for details on high pass filtering
@implementation HighpassFilter

-(id)initWithSampleRate:(double)rate cutoffFrequency:(double)freq
{
	self = [super init];
	if(self != nil)
	{
		double dt = 1.0 / rate;
		double RC = 1.0 / freq;
		filterConstant = RC / (dt + RC);
	}
	return self;
}

-(void)addAcceleration:(UIAcceleration*)accel
{
	double alpha = filterConstant;

	if(adaptive)
	{
		double d = Clamp(fabs(Norm(x, y, z) - Norm(accel.x, accel.y, accel.z)) / kAccelerometerMinStep - 1.0, 0.0, 1.0);
		alpha = d * filterConstant / kAccelerometerNoiseAttenuation + (1.0 - d) * filterConstant;
	}

	x = alpha * (x + accel.x - lastX);
	y = alpha * (y + accel.y - lastY);
	z = alpha * (z + accel.z - lastZ);

	lastX = accel.x;
	lastY = accel.y;
	lastZ = accel.z;
}

-(NSString*)name
{
	return adaptive ? @"Adaptive Highpass Filter" : @"Highpass Filter";
}

@end


text:
package com.bah.geterdun;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Logs an unrecoverable failure to process an event in the SLF4J logs.
 *
 */
public class NullFailureHandler<EVENT> implements FailureHandler<EVENT> {

  private static final Logger LOG = LoggerFactory
      .getLogger(NullFailureHandler.class);

  public void handleFailure(EVENT action) {
    LOG.error("Event permentantly failed: " + action);
  }

}


text:
package org.camunda.bpm.model.bpmn.instance;

/**
 * The BPMN Import element
 *
 * @author Daniel Meyer
 */
public interface Import extends BpmnModelElementInstance {

  String getNamespace();

  void setNamespace(String namespace);

  String getLocation();

  void setLocation(String location);

  String getImportType();

  void setImportType(String importType);

}