# Path: datapackage_pipelines/specs/specs.py
# SPEC_PARSERS = [
#     BasicPipelineParser(),
#     SourceSpecPipelineParser()
# ]
# YAML_LOADER = yaml.CLoader if 'CLoader' in yaml.__dict__ else yaml.Loader
# def resolve_processors(spec: PipelineSpec):
# def process_schedules(spec: PipelineSpec):
# def process_environment(spec: PipelineSpec):
# def find_specs(root_dir='.') -> PipelineSpec:
# def pipelines(prefixes=None, ignore_missing_deps=False, root_dir='.', status_manager=None):
# def register_all_pipelines(root_dir='.'):
# 
# Path: datapackage_pipelines/status/status_manager.py
# def status_mgr(root_dir='.') -> StatusManager:
#     global _status
#     global _root_dir
# 
#     if _status is not None and _root_dir == root_dir:
#         return _status
#     _root_dir = root_dir
#     _status = StatusManager(host=os.environ.get('DPP_REDIS_HOST'), root_dir=root_dir)
#     return _status
# 
# Path: datapackage_pipelines/utilities/stat_utils.py
# STATS_DPP_KEY = '.dpp'
# 
# STATS_OUT_DP_URL_KEY = 'out-datapackage-url'
# 
# Path: datapackage_pipelines/utilities/extended_json.py
# class json(object):
#     dumps = _dumps
#     loads = _loads
#     dumpl = _dumpl
#     loadl = _loadl
#     dump = _dump
#     load = _load
#     JSONDecodeError = _json.JSONDecodeError
# 
# Path: datapackage_pipelines/manager/runners/runner_config.py
# class RunnerConfiguration(object):
#     ENV_VAR = 'DPP_RUNNER_CONFIG'
#     DEFAULT_RUNNER_CONFIG = 'dpp-runners.yaml'
#     def __init__(self):
#     def get_runner_class(self, kind):
#     def get_runner(self, name):


import asyncio
import gzip
import logging
import os
from concurrent.futures import CancelledError
from json.decoder import JSONDecodeError
from ..specs.specs import resolve_executor
from ..status import status_mgr
from ..utilities.stat_utils import STATS_DPP_KEY, STATS_OUT_DP_URL_KEY
from ..utilities.extended_json import json
from .runners import runner_config

async def dequeue_errors(queue, out):
    while True:
        line = await queue.get()
        if line is None:
            break
        out.append(line)
        if len(out) > 1000:
            out.pop(0)


async def collect_stats(infile):
    reader = asyncio.StreamReader()
    reader_protocol = asyncio.StreamReaderProtocol(reader)
    transport, _ = await asyncio.get_event_loop() \
        .connect_read_pipe(lambda: reader_protocol, infile)
    count = 0
    dp = None
    stats = None
    while True:
        try:
            line = await reader.readline()
        except ValueError:
            logging.exception('Too large stats object!')
            break
        if line == b'':
            break
        stats = line
        if dp is None:
            try:

# Path: datapackage_pipelines/specs/specs.py
# SPEC_PARSERS = [
#     BasicPipelineParser(),
#     SourceSpecPipelineParser()
# ]
# YAML_LOADER = yaml.CLoader if 'CLoader' in yaml.__dict__ else yaml.Loader
# def resolve_processors(spec: PipelineSpec):
# def process_schedules(spec: PipelineSpec):
# def process_environment(spec: PipelineSpec):
# def find_specs(root_dir='.') -> PipelineSpec:
# def pipelines(prefixes=None, ignore_missing_deps=False, root_dir='.', status_manager=None):
# def register_all_pipelines(root_dir='.'):
# 
# Path: datapackage_pipelines/status/status_manager.py
# def status_mgr(root_dir='.') -> StatusManager:
#     global _status
#     global _root_dir
# 
#     if _status is not None and _root_dir == root_dir:
#         return _status
#     _root_dir = root_dir
#     _status = StatusManager(host=os.environ.get('DPP_REDIS_HOST'), root_dir=root_dir)
#     return _status
# 
# Path: datapackage_pipelines/utilities/stat_utils.py
# STATS_DPP_KEY = '.dpp'
# 
# STATS_OUT_DP_URL_KEY = 'out-datapackage-url'
# 
# Path: datapackage_pipelines/utilities/extended_json.py
# class json(object):
#     dumps = _dumps
#     loads = _loads
#     dumpl = _dumpl
#     loadl = _loadl
#     dump = _dump
#     load = _load
#     JSONDecodeError = _json.JSONDecodeError
# 
# Path: datapackage_pipelines/manager/runners/runner_config.py
# class RunnerConfiguration(object):
#     ENV_VAR = 'DPP_RUNNER_CONFIG'
#     DEFAULT_RUNNER_CONFIG = 'dpp-runners.yaml'
#     def __init__(self):
#     def get_runner_class(self, kind):
#     def get_runner(self, name):


# Path: datapackage_pipelines/manager/tasks.py
import asyncio
import gzip
import logging
import os
from concurrent.futures import CancelledError
from json.decoder import JSONDecodeError
from ..specs.specs import resolve_executor
from ..status import status_mgr
from ..utilities.stat_utils import STATS_DPP_KEY, STATS_OUT_DP_URL_KEY
from ..utilities.extended_json import json
from .runners import runner_config


async def dequeue_errors(queue, out):
    while True:
        line = await queue.get()
        if line is None:
            break
        out.append(line)
        if len(out) > 1000:
            out.pop(0)


async def collect_stats(infile):
    reader = asyncio.StreamReader()
    reader_protocol = asyncio.StreamReaderProtocol(reader)
    transport, _ = await asyncio.get_event_loop() \
        .connect_read_pipe(lambda: reader_protocol, infile)
    count = 0
    dp = None
    stats = None
    while True:
        try:
            line = await reader.readline()
        except ValueError:
            logging.exception('Too large stats object!')
            break
        if line == b'':
            break
        stats = line
        if dp is None:
            try:

                dp = json.loads(line.decode('ascii'))