# Path: datapackage_pipelines/wrapper/wrapper.py
# def ingest(debug=False):
#     params, datapackage, resource_iterator = _ingest(debug=debug)
#     return ProcessorContext(params, datapackage, resource_iterator)
# 
# def spew(dp, resources_iterator, stats=None, finalizer=None):
#     files = [stdout]
# 
#     cache_filename = ''
#     if len(cache) > 0:
#         if not os.path.exists('.cache'):
#             os.mkdir('.cache')
#         cache_filename = os.path.join('.cache', cache)
#         files.append(gzip.open(cache_filename+'.ongoing', 'wt'))
# 
#     expected_resources = \
#         len(list(filter(streaming, dp.get('resources', []))))
#     row_count = 0
#     try:
#         for f in files:
#             f.write(json.dumps(dp, sort_keys=True, ensure_ascii=True)+'\n')
#             f.flush()
#         num_resources = 0
#         for res in resources_iterator:
#             if hasattr(res, 'it') and res.it is None:
#                 continue
#             num_resources += 1
#             for f in files:
#                 f.write('\n')
#             try:
#                 for rec in res:
#                     try:
#                         line = json.dumpl(rec,
#                                         sort_keys=True,
#                                         ensure_ascii=True)
#                     except TypeError as e:
#                         logging.error('Failed to encode row to JSON: %s\nOffending row: %r', e, rec)
#                         raise
#                     # logging.error('SPEWING: {}'.format(line))
#                     for f in files:
#                         f.write(line+'\n')
#                     # logging.error('WROTE')
#                     row_count += 1
#             except CastError as e:
#                 for err in e.errors:
#                     logging.error('Failed to cast row: %s', err)
#                 raise
#         if num_resources != expected_resources:
#             logging.error('Expected to see %d resource(s) but spewed %d',
#                           expected_resources, num_resources)
#             assert num_resources == expected_resources
# 
#         aggregated_stats = {}
#         if not first:
#             stats_line = sys.stdin.readline().strip()
#             if len(stats_line) > 0:
#                 try:
#                     aggregated_stats = json.loads(stats_line)
#                     assert aggregated_stats is None or \
#                         isinstance(aggregated_stats, dict)
#                 except json.JSONDecodeError:
#                     logging.error('Failed to parse stats: %r', stats_line)
#         if stats is not None:
#             aggregated_stats.update(stats)
#         stats_json = json.dumps(aggregated_stats,
#                                 sort_keys=True,
#                                 ensure_ascii=True)
#         for f in files:
#             f.write('\n'+stats_json)
# 
#     except BrokenPipeError:
#         logging.error('Output pipe disappeared!')
#         sys.stderr.flush()
#         sys.exit(1)
# 
#     stdout.flush()
#     if row_count > 0:
#         logging.info('Processed %d rows', row_count)
# 
#     if finalizer is not None:
#         finalizer()
# 
#     for f in files:
#         f.write('\n')  # Signal to other processors that we're done
#         if f == stdout:
#             # Can't close sys.stdout, otherwise any subsequent
#             # call to print() will throw an exception
#             f.flush()
#         else:
#             f.close()
# 
#     if len(cache) > 0:
#         os.rename(cache_filename+'.ongoing', cache_filename)


import sys
from datapackage_pipelines.wrapper import ingest, spew


SINK_MAGIC = '>>> PROCESSED ROWS: '


def sink(res_iter_):
    count = 0
    for res in res_iter_:
        for row in res:
            count += 1
            if count % 100 == 0:
                sys.stderr.write('%s%d\n' % (SINK_MAGIC, count))
                sys.stderr.flush()
    sys.stderr.write('%s%d\n' % (SINK_MAGIC, count))
    sys.stderr.flush()
    yield from ()


if __name__ == '__main__':
    sys.stderr.write('%s%d\n' % (SINK_MAGIC, 0))
    sys.stderr.flush()
    params, dp, res_iter = ingest()

# Path: datapackage_pipelines/wrapper/wrapper.py
# def ingest(debug=False):
#     params, datapackage, resource_iterator = _ingest(debug=debug)
#     return ProcessorContext(params, datapackage, resource_iterator)
# 
# def spew(dp, resources_iterator, stats=None, finalizer=None):
#     files = [stdout]
# 
#     cache_filename = ''
#     if len(cache) > 0:
#         if not os.path.exists('.cache'):
#             os.mkdir('.cache')
#         cache_filename = os.path.join('.cache', cache)
#         files.append(gzip.open(cache_filename+'.ongoing', 'wt'))
# 
#     expected_resources = \
#         len(list(filter(streaming, dp.get('resources', []))))
#     row_count = 0
#     try:
#         for f in files:
#             f.write(json.dumps(dp, sort_keys=True, ensure_ascii=True)+'\n')
#             f.flush()
#         num_resources = 0
#         for res in resources_iterator:
#             if hasattr(res, 'it') and res.it is None:
#                 continue
#             num_resources += 1
#             for f in files:
#                 f.write('\n')
#             try:
#                 for rec in res:
#                     try:
#                         line = json.dumpl(rec,
#                                         sort_keys=True,
#                                         ensure_ascii=True)
#                     except TypeError as e:
#                         logging.error('Failed to encode row to JSON: %s\nOffending row: %r', e, rec)
#                         raise
#                     # logging.error('SPEWING: {}'.format(line))
#                     for f in files:
#                         f.write(line+'\n')
#                     # logging.error('WROTE')
#                     row_count += 1
#             except CastError as e:
#                 for err in e.errors:
#                     logging.error('Failed to cast row: %s', err)
#                 raise
#         if num_resources != expected_resources:
#             logging.error('Expected to see %d resource(s) but spewed %d',
#                           expected_resources, num_resources)
#             assert num_resources == expected_resources
# 
#         aggregated_stats = {}
#         if not first:
#             stats_line = sys.stdin.readline().strip()
#             if len(stats_line) > 0:
#                 try:
#                     aggregated_stats = json.loads(stats_line)
#                     assert aggregated_stats is None or \
#                         isinstance(aggregated_stats, dict)
#                 except json.JSONDecodeError:
#                     logging.error('Failed to parse stats: %r', stats_line)
#         if stats is not None:
#             aggregated_stats.update(stats)
#         stats_json = json.dumps(aggregated_stats,
#                                 sort_keys=True,
#                                 ensure_ascii=True)
#         for f in files:
#             f.write('\n'+stats_json)
# 
#     except BrokenPipeError:
#         logging.error('Output pipe disappeared!')
#         sys.stderr.flush()
#         sys.exit(1)
# 
#     stdout.flush()
#     if row_count > 0:
#         logging.info('Processed %d rows', row_count)
# 
#     if finalizer is not None:
#         finalizer()
# 
#     for f in files:
#         f.write('\n')  # Signal to other processors that we're done
#         if f == stdout:
#             # Can't close sys.stdout, otherwise any subsequent
#             # call to print() will throw an exception
#             f.flush()
#         else:
#             f.close()
# 
#     if len(cache) > 0:
#         os.rename(cache_filename+'.ongoing', cache_filename)


# Path: datapackage_pipelines/lib/internal/sink.py
import sys
from datapackage_pipelines.wrapper import ingest, spew



SINK_MAGIC = '>>> PROCESSED ROWS: '


def sink(res_iter_):
    count = 0
    for res in res_iter_:
        for row in res:
            count += 1
            if count % 100 == 0:
                sys.stderr.write('%s%d\n' % (SINK_MAGIC, count))
                sys.stderr.flush()
    sys.stderr.write('%s%d\n' % (SINK_MAGIC, count))
    sys.stderr.flush()
    yield from ()


if __name__ == '__main__':
    sys.stderr.write('%s%d\n' % (SINK_MAGIC, 0))
    sys.stderr.flush()
    params, dp, res_iter = ingest()

    spew({'name': 'boop', 'resources': []},