#include #include #include #include "lib/mlrutil.h" #include "lib/mlr_globals.h" #include "containers/lrec.h" #include "containers/sllv.h" #include "input/lrec_readers.h" #include "mapping/mappers.h" #include "output/lrec_writers.h" static int do_stream_chained_in_place(context_t* pctx, cli_opts_t* popts); static int do_stream_chained_to_stdout(context_t* pctx, sllv_t* pmapper_list, cli_opts_t* popts); static int do_file_chained(char* filename, context_t* pctx, lrec_reader_t* plrec_reader, sllv_t* pmapper_list, lrec_writer_t* plrec_writer, FILE* output_stream, cli_opts_t* popts); static sllv_t* chain_map(lrec_t* pinrec, context_t* pctx, sllve_t* pmapper_list_head); static void drive_lrec(lrec_t* pinrec, context_t* pctx, sllve_t* pmapper_list_head, lrec_writer_t* plrec_writer, FILE* output_stream); typedef void progress_indicator_t(context_t* pctx, long long nr_progress_mod); static void null_progress_indicator(context_t* pctx, long long nr_progress_mod); static void stderr_progress_indicator(context_t* pctx, long long nr_progress_mod); // ---------------------------------------------------------------- int do_stream_chained(context_t* pctx, sllv_t* pmapper_list, cli_opts_t* popts) { if (popts->do_in_place) { return do_stream_chained_in_place(pctx, popts); } else { return do_stream_chained_to_stdout(pctx, pmapper_list, popts); } } // ---------------------------------------------------------------- // For in-place mode, reconstruct the mappers on each input file. E.g. 'mlr -I head -n 2 // foo bar' should do head -n 2 on foo as well as on bar. // // I could have implemented this with a single construction of the mappers and having each // mapper implement a virtual reset() method. However, having effectively two initalizers // per mapper -- constructor and reset method -- I'd surely miss some logic somewhere. // With in-place mode being a less frequently used code path, this would likely lead to // latent bugs. So while it seems inelegant to re-parse part of the command line over and // over again to reconstruct mappers for each file, this approach leads to greater code // stability. static int do_stream_chained_in_place(context_t* pctx, cli_opts_t* popts) { MLR_INTERNAL_CODING_ERROR_IF(popts->filenames == NULL); MLR_INTERNAL_CODING_ERROR_IF(popts->filenames->length == 0); int ok = 1; // Read from each file name in turn for (sllse_t* pe = popts->filenames->phead; pe != NULL; pe = pe->pnext) { // Allocate reader, mappers, and writer individually for each file name. // This way CSV headers appear in each file, head -n 10 puts 10 rows for // each output file, and so on. lrec_reader_t* plrec_reader = lrec_reader_alloc_or_die(&popts->reader_opts); lrec_writer_t* plrec_writer = lrec_writer_alloc_or_die(&popts->writer_opts); // Note that the command-line parsers can operate destructively on argv, // e.g. verbs which take comma-delimited field names splitting on commas. // For this reason we need to duplicate argv on each run. We need to free // after processing in case mappers have retained pointers into argv. int argi = popts->mapper_argb; char** argv_copy = copy_argv(popts->original_argv); sllv_t* pmapper_list = cli_parse_mappers(argv_copy, &argi, popts->argc, popts); // Should not have been allowed by the CLI parser: MLR_INTERNAL_CODING_ERROR_IF(pmapper_list->length < 1); char* filename = pe->value; char* tempname = alloc_suffixed_temp_file_name(filename); FILE* output_stream = fopen(tempname, "wb"); if (output_stream == NULL) { perror("fopen"); fprintf(stderr, "%s: Could not open \"%s\" for write.\n", MLR_GLOBALS.bargv0, tempname); exit(1); } pctx->filenum++; pctx->filename = filename; pctx->fnr = 0; ok = do_file_chained(filename, pctx, plrec_reader, pmapper_list, plrec_writer, output_stream, popts) && ok; // For in-place mode, there's no breaking from the loop over input files. Just an early // return from the mapper chain, which has already just happened. if (pctx->force_eof == TRUE) // e.g. mlr head pctx->force_eof = FALSE; // Mappers and writers receive end-of-stream notifications via null input record. // Do that, now that data from the input file have been exhausted. drive_lrec(NULL, pctx, pmapper_list->phead, plrec_writer, output_stream); // Drain the pretty-printer. plrec_writer->pprocess_func(plrec_writer->pvstate, output_stream, NULL, pctx); fclose(output_stream); int rc = rename(tempname, filename); if (rc != 0) { perror("rename"); fprintf(stderr, "%s: Could not rename \"%s\" to \"%s\".\n", MLR_GLOBALS.bargv0, tempname, filename); exit(1); } free(tempname); plrec_reader->pfree_func(plrec_reader); plrec_writer->pfree_func(plrec_writer, pctx); mapper_chain_free(pmapper_list, pctx); free_argv_copy(argv_copy); } return ok; } // ---------------------------------------------------------------- static int do_stream_chained_to_stdout(context_t* pctx, sllv_t* pmapper_list, cli_opts_t* popts) { FILE* output_stream = stdout; lrec_reader_t* plrec_reader = lrec_reader_alloc_or_die(&popts->reader_opts); lrec_writer_t* plrec_writer = lrec_writer_alloc_or_die(&popts->writer_opts); MLR_INTERNAL_CODING_ERROR_IF(pmapper_list->length < 1); // Should not have been allowed by the CLI parser. int ok = 1; if (popts->filenames == NULL) { // No input at all } else if (popts->filenames->length == 0) { // Zero file names means read from standard input pctx->filenum++; pctx->filename = "(stdin)"; pctx->fnr = 0; ok = do_file_chained("-", pctx, plrec_reader, pmapper_list, plrec_writer, output_stream, popts) && ok; } else { // Read from each file name in turn for (sllse_t* pe = popts->filenames->phead; pe != NULL; pe = pe->pnext) { char* filename = pe->value; pctx->filenum++; pctx->filename = filename; pctx->fnr = 0; ok = do_file_chained(filename, pctx, plrec_reader, pmapper_list, plrec_writer, output_stream, popts) && ok; if (pctx->force_eof == TRUE) // e.g. mlr head break; } } // Mappers and writers receive end-of-stream notifications via null input record. // Do that, now that data from all input file(s) have been exhausted. drive_lrec(NULL, pctx, pmapper_list->phead, plrec_writer, output_stream); // Drain the pretty-printer. plrec_writer->pprocess_func(plrec_writer->pvstate, output_stream, NULL, pctx); plrec_reader->pfree_func(plrec_reader); plrec_writer->pfree_func(plrec_writer, pctx); return ok; } // ---------------------------------------------------------------- static int do_file_chained(char* filename, context_t* pctx, lrec_reader_t* plrec_reader, sllv_t* pmapper_list, lrec_writer_t* plrec_writer, FILE* output_stream, cli_opts_t* popts) { void* pvhandle = plrec_reader->popen_func(plrec_reader->pvstate, popts->reader_opts.prepipe, filename); progress_indicator_t* pindicator = popts->nr_progress_mod == 0LL ? null_progress_indicator : stderr_progress_indicator; // Start-of-file hook, e.g. expecting CSV headers on input. plrec_reader->psof_func(plrec_reader->pvstate, pvhandle); while (1) { lrec_t* pinrec = plrec_reader->pprocess_func(plrec_reader->pvstate, pvhandle, pctx); if (pinrec == NULL) break; if (pctx->force_eof == TRUE) { // e.g. mlr head lrec_free(pinrec); break; } pctx->nr++; pctx->fnr++; pindicator(pctx, popts->nr_progress_mod); drive_lrec(pinrec, pctx, pmapper_list->phead, plrec_writer, output_stream); } plrec_reader->pclose_func(plrec_reader->pvstate, pvhandle, popts->reader_opts.prepipe); return 1; } // ---------------------------------------------------------------- static void drive_lrec(lrec_t* pinrec, context_t* pctx, sllve_t* pmapper_list_head, lrec_writer_t* plrec_writer, FILE* output_stream) { sllv_t* outrecs = chain_map(pinrec, pctx, pmapper_list_head); if (outrecs != NULL) { for (sllve_t* pe = outrecs->phead; pe != NULL; pe = pe->pnext) { lrec_t* poutrec = pe->pvvalue; if (poutrec != NULL) // writer frees records (sllv void-star payload) plrec_writer->pprocess_func(plrec_writer->pvstate, output_stream, poutrec, pctx); } sllv_free(outrecs); // we free the list } } // ---------------------------------------------------------------- // Map a single input record (maybe null at end of input stream) to zero or // more output records. // // Return: list of lrec_t*. Input: lrec_t* and list of mapper_t*. static sllv_t* chain_map(lrec_t* pinrec, context_t* pctx, sllve_t* pmapper_list_head) { mapper_t* pmapper = pmapper_list_head->pvvalue; sllv_t* outrecs = pmapper->pprocess_func(pinrec, pctx, pmapper->pvstate); if (pmapper_list_head->pnext == NULL) { return outrecs; } else if (outrecs == NULL) { // end of input stream return NULL; } else { sllv_t* nextrecs = sllv_alloc(); for (sllve_t* pe = outrecs->phead; pe != NULL; pe = pe->pnext) { lrec_t* poutrec = pe->pvvalue; sllv_t* nextrecsi = chain_map(poutrec, pctx, pmapper_list_head->pnext); sllv_transfer(nextrecs, nextrecsi); sllv_free(nextrecsi); } sllv_free(outrecs); return nextrecs; } } // ---------------------------------------------------------------- static void stderr_progress_indicator(context_t* pctx, long long nr_progress_mod) { long long remainder = pctx->nr % nr_progress_mod; if (remainder == 0) { fprintf(stderr, "NR=%lld FNR=%lld FILENAME=%s\n", pctx->nr, pctx->fnr, pctx->filename); } } static void null_progress_indicator(context_t* pctx, long long nr_progress_mod) { }