Job TAPI usage scenarios

Run and wait

Run a program
Log all stdout and stderr via the Logger
Wait for the program termination

tapi_job_factory_t *factory = NULL;
tapi_job_t *job = NULL
tapi_job_channel_t *out_channels[2];

CHECK_RC(tapi_job_factory_rpc_create(pco, &factory));
CHECK_RC(tapi_job_create(factory, NULL, "/usr/bin/tool",
                         (const char *[]){"tool", "arg1", "arg2",
                         NULL}, NULL, &job));
CHECK_RC(tapi_job_alloc_output_channels(job, 2, out_channels));
CHECK_RC(tapi_job_attach_filter(TAPI_JOB_CHANNEL_SET(out_channels[0],
                                                     out_channels[1]),
                                "Filter", false, TE_LL_INFO, NULL));
CHECK_RC(tapi_job_start(job));
...
CHECK_RC(tapi_job_wait(job, -1, &status));
if (status.type != TAPI_JOB_STATUS_EXITED || status.value != 0)
   TEST_FAIL("Tool failed");
CHECK_RC(tapi_job_destroy(job, -1));
tapi_job_factory_destroy(factory);

Run and read

Run a program
Read all the stdout while logging all the stderr
When all the data are read, the program is assumed to terminate

tapi_job_factory_t *factory = NULL;
tapi_job_t *job = NULL;
tapi_job_buffer_t buf = TAPI_JOB_BUFFER_INIT;
tapi_job_channel_t *out_channels[2];
tapi_job_channel_t *out_filter;

CHECK_RC(tapi_job_factory_rpc_create(pco, &factory));
CHECK_RC(tapi_job_create(factory, NULL, "/usr/bin/tool",
                         (const char *[]){"tool", "arg1", "arg2",
                         NULL}, NULL, &job));
CHECK_RC(tapi_job_alloc_output_channels(job, 2, out_channels));
CHECK_RC(tapi_job_attach_filter(TAPI_JOB_CHANNEL_SET(out_channels[0]),
                                "Out filter", true, 0, &out_filter));
CHECK_RC(tapi_job_attach_filter(TAPI_JOB_CHANNEL_SET(out_channels[1]),
                                "Error filter", false, TE_LL_INFO, NULL));
CHECK_RC(tapi_job_start(job));
while (!buf.eos)
{
    te_string_reset(&buf.data);
    CHECK_RC(tapi_job_receive(TAPI_JOB_CHANNEL_SET(out_filter), -1, &buf));
    ...
}
CHECK_RC(tapi_job_wait(job, 0, &status));
if (status.type != TAPI_JOB_STATUS_EXITED || status.value != 0)
   TEST_FAIL("Tool failed");
CHECK_RC(tapi_job_destroy(job, -1));
tapi_job_factory_destroy(factory);

Run and wait for

Run a program
Log the stderr line matching ERROR as TE errors
Wait for the string Completed at the stdout

tapi_job_factory_t *factory = NULL;
tapi_job_t *job = NULL;
tapi_job_channel_t *out_channels[2];
tapi_job_channel_t *out_filters[2];

CHECK_RC(tapi_job_factory_rpc_create(pco, &factory));
CHECK_RC(tapi_job_create(factory, NULL, "/usr/bin/tool",
                         (const char *[]){"tool", "arg1", "arg2",
                         NULL}, NULL, &job));
CHECK_RC(tapi_job_alloc_output_channels(job, 2, out_channels));
CHECK_RC(tapi_job_attach_filter(TAPI_JOB_CHANNEL_SET(out_channels[0]),
                                "Out filter", true, 0, &out_filters[0]));
CHECK_RC(tapi_job_filter_add_regexp(out_filters[0], "Completed", 0));
CHECK_RC(tapi_job_attach_filter(TAPI_JOB_CHANNEL_SET(out_channels[1]),
                                "Error filter", false, TE_LL_ERROR,
                                &out_filters[1]));
CHECK_RC(tapi_job_filter_add_regexp(out_filters[1],
                                   "^ERROR:(.*)$", 1);

CHECK_RC(tapi_job_start(job));

CHECK_RC(tapi_job_poll(TAPI_JOB_CHANNEL_SET(out_filters[0]), -1));

CHECK_RC(tapi_job_wait(job, 0, &status));
if (status.type != TAPI_JOB_STATUS_EXITED || status.value != 0)
   TEST_FAIL("Tool failed");
CHECK_RC(tapi_job_destroy(job, -1));
tapi_job_factory_destroy(factory);

Run an interactive program

Run an interactive program via a pseudo-terminal
Wait for the prompt
Send a command

Repeat the previous steps, logging all the output until the program terminates

tapi_job_factory_t *factory = NULL;
tapi_job_t *job = NULL;
tapi_job_channel_t *in_channel;
tapi_job_channel_t *out_channel;
tapi_job_channel_t *prompt_filter;

CHECK_RC(tapi_job_factory_rpc_create(pco, &factory));
CHECK_RC(tapi_job_create(factory, "pty_spawner", "/usr/bin/tool",
                         (const char *[]){"tool", "arg1", "arg2",
                         NULL}, NULL, &job));
CHECK_RC(tapi_job_alloc_output_channels(job, 1, &out_channel));
CHECK_RC(tapi_job_alloc_input_channels(job, 1, &in_channel));
CHECK_RC(tapi_job_attach_filter(TAPI_JOB_CHANNEL_SET(out_channel),
                                "Readable filter", true, 0, &prompt_filter));
CHECK_RC(tapi_job_filter_add_regexp(prompt_filter, "^\\$ ", 0));
CHECK_RC(tapi_job_attach_filter(TAPI_JOB_CHANNEL_SET(out_channel),
                                "Log filter", false, TE_LL_RING, NULL));

CHECK_RC(tapi_job_start(job));

while (tapi_job_wait(job, 0, &status) == TE_EINPROGRESS)
{
    CHECK_RC(tapi_job_receive(TAPI_JOB_CHANNEL_SET(prompt_filter), -1, NULL));
    CHECK_RC(tapi_job_send(job, in_channel, &command_string));
}

if (status.type != TAPI_JOB_STATUS_EXITED || status.value != 0)
   TEST_FAIL("Tool failed");
CHECK_RC(tapi_job_destroy(job, -1));
tapi_job_factory_destroy(factory);

Terminate a job and if it hangs, forcedly kill it

...
CHECK_RC(tapi_job_kill(job, SIGTERM));
if (tapi_job_wait(job, TIMEOUT, &status) == TE_INPROGRESS)
{
    CHECK_RC(tapi_job_kill(job, SIGKILL));
    CHECK_RC(tapi_job_wait(job, 0, &status));
}
...

Start a job several times and get its data

unsigned int eos_count;
tapi_job_buffer buf = TAPI_JOB_BUFFER_INIT;
tapi_job_factory_t *factory = NULL;
tapi_job_t *job = NULL;
tapi_job_channel_t *out_channel;
tapi_job_channel_t *out_filter;
tapi_job_channel_t *out_filter2;

CHECK_RC(tapi_job_factory_rpc_create(pco, &factory));
CHECK_RC(tapi_job_create(factory, "pty_spawner", "/usr/bin/tool",
                         (const char *[]){"tool", "arg1", "arg2",
                         NULL}, NULL, &job));
CHECK_RC(tapi_job_alloc_output_channels(job, 1, &out_channel));
CHECK_RC(tapi_job_attach_filter(TAPI_JOB_CHANNEL_SET(out_channel),
                                "Readable filter", true, 0, &out_filter));
CHECK_RC(tapi_job_attach_filter(TAPI_JOB_CHANNEL_SET(out_channel),
                                "Readable filter2", true, 0, &out_filter2));

CHECK_RC(tapi_job_start(job));
CHECK_RC(tapi_job_wait(job, BIG_TIMEOUT, NULL));

// Get data from the first run from both filters
eos_count = 0;
while (eos_count < 2)
{
    te_string_reset(&buf.data);
    CHECK_RC(tapi_job_receive(TAPI_JOB_CHANNEL_SET(out_filter, out_filter2),
                              -1, &buf));
    if (buf.eos)
        eos_count++;
    ...
}

CHECK_RC(tapi_job_start(job));
CHECK_RC(tapi_job_wait(job, BIG_TIMEOUT, NULL));

// Get data from the second run from both filters
eos_count = 0;
while (eos_count < 2)
{
    te_string_reset(&buf.data);
    CHECK_RC(tapi_job_receive(TAPI_JOB_CHANNEL_SET(out_filter, out_filter2),
                              -1, &buf));
    if (buf.eos)
        eos_count++;
    ...
}

CHECK_RC(tapi_job_destroy(job, -1));
tapi_job_factory_destroy(factory);