Mirrors/miller
1
0
Fork 0
mirror of https://github.com/johnkerl/miller.git synced 2026-08-02 04:22:59 +00:00
Code Issues Projects Releases Wiki Activity

pipe-out iterate

Browse source
This commit is contained in:
John Kerl 2016-08-04 08:56:15 -04:00
parent cefcf7ed97
commit 69a05a32bb
5 changed files with 100 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

5
c/output/file_output_mode.h
View file

@ -2,8 +2,9 @@
#define FILE_OUTPUT_MODE_H
typedef enum _file_output_mode_t {
MODE_WRITE,
MODE_APPEND,
MODE_WRITE, // fopen/fclose
MODE_APPEND, // fopen/fclose
MODE_PIPE, // popen/pclose
} file_output_mode_t;
static inline char* get_mode_string(file_output_mode_t file_output_mode) {

69
c/output/multi_lrec_writer.c
View file

@ -18,7 +18,7 @@ void multi_lrec_writer_free(multi_lrec_writer_t* pmlw) {
for (lhmsve_t* pe = pmlw->pnames_to_lrec_writers_and_fps->phead; pe != NULL; pe = pe->pnext) {
lrec_writer_and_fp_t* pstate = pe->pvvalue;
pstate->plrec_writer->pfree_func(pstate->plrec_writer);
free(pstate->filename);
free(pstate->filename_or_command);
}
lhmsv_free(pmlw->pnames_to_lrec_writers_and_fps);
@ -26,10 +26,10 @@ void multi_lrec_writer_free(multi_lrec_writer_t* pmlw) {
}
// ----------------------------------------------------------------
void multi_lrec_writer_output_srec(multi_lrec_writer_t* pmlw, lrec_t* poutrec, char* filename,
void multi_lrec_writer_output_srec(multi_lrec_writer_t* pmlw, lrec_t* poutrec, char* filename_or_command,
file_output_mode_t file_output_mode, int flush_every_record)
{
lrec_writer_and_fp_t* pstate = lhmsv_get(pmlw->pnames_to_lrec_writers_and_fps, filename);
lrec_writer_and_fp_t* pstate = lhmsv_get(pmlw->pnames_to_lrec_writers_and_fps, filename_or_command);
if (pstate == NULL) {
pstate = mlr_malloc_or_die(sizeof(lrec_writer_and_fp_t));
cli_opts_t* popts = MLR_GLOBALS.popts;
@ -42,17 +42,30 @@ void multi_lrec_writer_output_srec(multi_lrec_writer_t* pmlw, lrec_t* poutrec, c
MLR_GLOBALS.bargv0, __FILE__, __LINE__);
exit(1);
}
pstate->filename = mlr_strdup_or_die(filename);
pstate->filename_or_command = mlr_strdup_or_die(filename_or_command);
char* mode_string = get_mode_string(file_output_mode);
char* mode_desc = get_mode_desc(file_output_mode);
pstate->output_stream = fopen(filename, mode_string);
if (pstate->output_stream == NULL) {
perror("fopen");
fprintf(stderr, "%s: failed fopen for %s on \"%s\".\n", MLR_GLOBALS.bargv0, mode_desc, filename);
exit(1);
if (file_output_mode == MODE_PIPE) {
pstate->is_popen = TRUE;
pstate->output_stream = popen(filename_or_command, mode_string);
if (pstate->output_stream == NULL) {
perror("popen");
fprintf(stderr, "%s: failed popen for %s on \"%s\".\n",
MLR_GLOBALS.bargv0, mode_desc, filename_or_command);
exit(1);
}
} else {
pstate->is_popen = FALSE;
pstate->output_stream = fopen(filename_or_command, mode_string);
if (pstate->output_stream == NULL) {
perror("fopen");
fprintf(stderr, "%s: failed fopen for %s on \"%s\".\n",
MLR_GLOBALS.bargv0, mode_desc, filename_or_command);
exit(1);
}
}
lhmsv_put(pmlw->pnames_to_lrec_writers_and_fps, mlr_strdup_or_die(filename), pstate, FREE_ENTRY_KEY);
lhmsv_put(pmlw->pnames_to_lrec_writers_and_fps, mlr_strdup_or_die(filename_or_command), pstate, FREE_ENTRY_KEY);
}
pstate->plrec_writer->pprocess_func(pstate->output_stream, poutrec, pstate->plrec_writer->pvstate);
@ -61,16 +74,24 @@ void multi_lrec_writer_output_srec(multi_lrec_writer_t* pmlw, lrec_t* poutrec, c
if (flush_every_record)
fflush(pstate->output_stream);
} else {
if (fclose(pstate->output_stream) != 0) {
perror("fclose");
fprintf(stderr, "%s: fclose error on \"%s\".\n", MLR_GLOBALS.bargv0, filename);
exit(1);
if (pstate->is_popen) {
if (pclose(pstate->output_stream) != 0) {
perror("pclose");
fprintf(stderr, "%s: pclose error on \"%s\".\n", MLR_GLOBALS.bargv0, filename_or_command);
exit(1);
}
} else {
if (fclose(pstate->output_stream) != 0) {
perror("fclose");
fprintf(stderr, "%s: fclose error on \"%s\".\n", MLR_GLOBALS.bargv0, filename_or_command);
exit(1);
}
}
pstate->output_stream = NULL;
}
}
void multi_lrec_writer_output_list(multi_lrec_writer_t* pmlw, sllv_t* poutrecs, char* filename,
void multi_lrec_writer_output_list(multi_lrec_writer_t* pmlw, sllv_t* poutrecs, char* filename_or_command,
file_output_mode_t file_output_mode, int flush_every_record)
{
if (poutrecs == NULL) // synonym for empty record-list
@ -78,7 +99,7 @@ void multi_lrec_writer_output_list(multi_lrec_writer_t* pmlw, sllv_t* poutrecs,
while (poutrecs->phead) {
lrec_t* poutrec = sllv_pop(poutrecs);
multi_lrec_writer_output_srec(pmlw, poutrec, filename, file_output_mode, flush_every_record);
multi_lrec_writer_output_srec(pmlw, poutrec, filename_or_command, file_output_mode, flush_every_record);
}
}
@ -87,10 +108,18 @@ void multi_lrec_writer_drain(multi_lrec_writer_t* pmlw) {
lrec_writer_and_fp_t* pstate = pe->pvvalue;
pstate->plrec_writer->pprocess_func(pstate->output_stream, NULL, pstate->plrec_writer->pvstate);
fflush(pstate->output_stream);
if (fclose(pstate->output_stream) != 0) {
perror("fclose");
fprintf(stderr, "%s: fclose error on \"%s\".\n", MLR_GLOBALS.bargv0, pstate->filename);
exit(1);
if (pstate->is_popen) {
if (pclose(pstate->output_stream) != 0) {
perror("pclose");
fprintf(stderr, "%s: pclose error on \"%s\".\n", MLR_GLOBALS.bargv0, pstate->filename_or_command);
exit(1);
}
} else {
if (fclose(pstate->output_stream) != 0) {
perror("fclose");
fprintf(stderr, "%s: fclose error on \"%s\".\n", MLR_GLOBALS.bargv0, pstate->filename_or_command);
exit(1);
}
}
}
}

12
c/output/multi_lrec_writer.h
View file

@ -8,25 +8,29 @@
#include "output/file_output_mode.h"
// ----------------------------------------------------------------
// This is the value struct for the hashmap:
typedef struct _lrec_writer_and_fp_t {
lrec_writer_t* plrec_writer;
char* filename;
char* filename_or_command;
FILE* output_stream;
int is_popen;
} lrec_writer_and_fp_t;
typedef struct _multi_lrec_writer_t {
// xxx to do: bound the number of open files and LRU them.
lhmsv_t* pnames_to_lrec_writers_and_fps;
} multi_lrec_writer_t;
// ----------------------------------------------------------------
multi_lrec_writer_t* multi_lrec_writer_alloc();
void multi_lrec_writer_free(multi_lrec_writer_t* pmlw);
void multi_lrec_writer_output_srec(multi_lrec_writer_t* pmlw, lrec_t* poutrec, char* filename,
void multi_lrec_writer_output_srec(multi_lrec_writer_t* pmlw, lrec_t* poutrec, char* filename_or_command,
file_output_mode_t file_output_mode, int flush_every_record);
void multi_lrec_writer_output_list(multi_lrec_writer_t* pmlw, sllv_t* poutrecs, char* filename,
void multi_lrec_writer_output_list(multi_lrec_writer_t* pmlw, sllv_t* poutrecs, char* filename_or_command,
file_output_mode_t file_output_mode, int flush_every_record);
void multi_lrec_writer_drain(multi_lrec_writer_t* pmlw);
#endif // MULTI_LREC_WRITER_H

45
c/output/multi_out.c
View file

@ -12,8 +12,12 @@ multi_out_t* multi_out_alloc() {
// ----------------------------------------------------------------
void multi_out_close(multi_out_t* pmo) {
for (lhmsve_t* pe = pmo->pnames_to_fps->phead; pe != NULL; pe = pe->pnext) {
FILE* fp = pe->pvvalue;
fclose(fp);
fp_and_flag_t* pstate = pe->pvvalue;
if (pstate->is_popen) {
pclose(pstate->output_stream);
} else {
fclose(pstate->output_stream);
}
}
}
@ -26,21 +30,32 @@ void multi_out_free(multi_out_t* pmo) {
}
// ----------------------------------------------------------------
FILE* multi_out_get(multi_out_t* pmo, char* filename, file_output_mode_t file_output_mode) {
FILE* outfp = lhmsv_get(pmo->pnames_to_fps, filename);
if (outfp == NULL) {
FILE* multi_out_get(multi_out_t* pmo, char* filename_or_command, file_output_mode_t file_output_mode) {
fp_and_flag_t* pstate = lhmsv_get(pmo->pnames_to_fps, filename_or_command);
if (pstate == NULL) {
pstate = mlr_malloc_or_die(sizeof(fp_and_flag_t));
char* mode_string = get_mode_string(file_output_mode);
char* mode_desc = get_mode_desc(file_output_mode);
outfp = fopen(filename, mode_string);
if (outfp == NULL) {
perror("fopen");
fprintf(stderr, "%s: failed fopen for %s of \"%s\".\n",
MLR_GLOBALS.bargv0, mode_desc, filename);
exit(1);
if (file_output_mode == MODE_PIPE) {
pstate->is_popen = TRUE;
pstate->output_stream = popen(filename_or_command, mode_string);
if (pstate->output_stream == NULL) {
perror("popen");
fprintf(stderr, "%s: failed popen for %s of \"%s\".\n",
MLR_GLOBALS.bargv0, mode_desc, filename_or_command);
exit(1);
}
} else {
pstate->is_popen = FALSE;
pstate->output_stream = fopen(filename_or_command, mode_string);
if (pstate->output_stream == NULL) {
perror("fopen");
fprintf(stderr, "%s: failed fopen for %s of \"%s\".\n",
MLR_GLOBALS.bargv0, mode_desc, filename_or_command);
exit(1);
}
}
lhmsv_put(pmo->pnames_to_fps, mlr_strdup_or_die(filename), outfp, FREE_ENTRY_KEY);
lhmsv_put(pmo->pnames_to_fps, mlr_strdup_or_die(filename_or_command), pstate, FREE_ENTRY_KEY);
}
return outfp;
return pstate->output_stream;
}
// xxx fcloses !

12
c/output/multi_out.h
View file

@ -6,15 +6,23 @@
#include "output/file_output_mode.h"
// ----------------------------------------------------------------
// This is the value struct for the hashmap:
typedef struct _fp_and_flag_t {
FILE* output_stream;
int is_popen;
} fp_and_flag_t;
typedef struct _multi_out_t {
// xxx to do: bound the number of open files and LRU them.
lhmsv_t* pnames_to_fps;
} multi_out_t;
// ----------------------------------------------------------------
multi_out_t* multi_out_alloc();
void multi_out_close(multi_out_t* pmo);
void multi_out_free(multi_out_t* pmo);
FILE* multi_out_get(multi_out_t* pmo, char* filename, file_output_mode_t file_output_mode);
FILE* multi_out_get(multi_out_t* pmo, char* filename_or_command, file_output_mode_t file_output_mode);
#endif // MULTI_OUT_H