23205294858

Committed 17 Mar 2026 04:36PM UTC coverage: 80.218% (+7.0%) from 73.258%

Build # 23205294858

Build Type

Pull #9

github

Committed by

web-flow

Commit Message

Merge 103bcc124 into 382e8a63e

Pull Request Pull Request #9: feat(poller): implemented linux/epoll support

Coverage Stats

107 of 132 new or added lines in 3 files covered. (81.06%)

2 existing lines in 1 file now uncovered.

442 of 551 relevant lines covered (80.22%)

7974.78 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.19

/src/reader.c

#include <assert.h>
#include <jp_poller.h>
#include <jp_queue.h>
#include <jp_reader.h>
#include <stdlib.h>
#include <unistd.h>

jp_errno_t jp_reader_consume(jp_reader_ctx_t ctx) {
    jp_errno_t err        = 0;
    ssize_t read_len      = 0;
    jp_block_t* block     = NULL;
    void* target_buffer   = NULL;
    unsigned char* buffer = malloc(ctx.chunk_size);
    jp_poller_t* poller   = jp_poller_create(100);

    if (buffer == NULL || poller == NULL) {
        err = JP_ENOMEMORY;
        goto clean_up;
    }

    err = jp_poller_poll(poller, ctx.input_stream);
    if (err) {
        goto clean_up;
    }

    while (true) {
        err = jp_poller_wait(poller);
        if (err == JP_EREAD_FAILED) {
            break;
        }
        if (err == JP_ETRYAGAIN) {
            continue;
        }
        while (true) {
            err = jp_queue_push_uncommitted(ctx.queue, &block);
            if (err == JP_ESHUTTING_DOWN) {
                goto clean_up;
            }
            target_buffer = err == JP_EMSG_SHOULD_DROP ? buffer : block->data;
            read_len      = read(ctx.input_stream, target_buffer, ctx.chunk_size);
            if (read_len == 0) {
                goto clean_up;
            }
            if (read_len < 0) {
                if (errno == EINTR) {
                    continue;
                }
                if (JP_ERRNO_EAGAIN(errno)) {
                    break;
                }
                err = JP_EREAD_FAILED;
                goto clean_up;
            }
            if (err == 0) {
                block->length = (size_t) read_len;
                jp_queue_push_commit(ctx.queue);
            }
        }
    }

clean_up:
    if (err == 0 || JP_QUEUE_IS_GRACEFUL_ERR(err)) {
        err = 0;
    } else {
        jp_errno_log_err(err);
    }
    JP_FREE(buffer);
    jp_poller_destroy(poller);
    jp_queue_finalize(ctx.queue);
    return err;
}

1	#include <assert.h>
2	#include <jp_poller.h>
3	#include <jp_queue.h>
4	#include <jp_reader.h>
5	#include <stdlib.h>
6	#include <unistd.h>
7
8	jp_errno_t jp_reader_consume(jp_reader_ctx_t ctx) {	10✔
9	jp_errno_t err = 0;	10✔
10	ssize_t read_len = 0;	10✔
11	jp_block_t* block = NULL;	10✔
12	void* target_buffer = NULL;	10✔
13	unsigned char* buffer = malloc(ctx.chunk_size);	10✔
14	jp_poller_t* poller = jp_poller_create(100);	10✔
15
16	if (buffer == NULL \|\| poller == NULL) {	10✔
NEW 17	err = JP_ENOMEMORY;	×
NEW 18	goto clean_up;	×
19	}
20
21	err = jp_poller_poll(poller, ctx.input_stream);	10✔
22	if (err) {	10✔
NEW 23	goto clean_up;	×
24	}
25
26	while (true) {	4,694✔
27	err = jp_poller_wait(poller);	4,694✔
28	if (err == JP_EREAD_FAILED) {	4,694✔
29	break;
30	}
31	if (err == JP_ETRYAGAIN) {	4,694✔
NEW 32	continue;	×
33	}
34	while (true) {	100,367✔
35	err = jp_queue_push_uncommitted(ctx.queue, &block);	100,367✔
36	if (err == JP_ESHUTTING_DOWN) {	100,367✔
NEW 37	goto clean_up;	×
38	}
39	target_buffer = err == JP_EMSG_SHOULD_DROP ? buffer : block->data;	100,367✔
40	read_len = read(ctx.input_stream, target_buffer, ctx.chunk_size);	100,367✔
41	if (read_len == 0) {	100,367✔
42	goto clean_up;	10✔
43	}
44	if (read_len < 0) {	100,357✔
45	if (errno == EINTR) {	4,684✔
NEW 46	continue;	×
47	}
48	if (JP_ERRNO_EAGAIN(errno)) {	4,684✔
49	break;
50	}
NEW 51	err = JP_EREAD_FAILED;	×
NEW 52	goto clean_up;	×
53	}
54	if (err == 0) {	95,673✔
55	block->length = (size_t) read_len;	95,661✔
56	jp_queue_push_commit(ctx.queue);	95,661✔
57	}
58	}
59	}
60
NEW 61	clean_up:	×
62	if (err == 0 \|\| JP_QUEUE_IS_GRACEFUL_ERR(err)) {	10✔
63	err = 0;
64	} else {
NEW 65	jp_errno_log_err(err);	×
66	}
67	JP_FREE(buffer);	10✔
68	jp_poller_destroy(poller);	10✔
69	jp_queue_finalize(ctx.queue);	10✔
70	return err;	10✔
71	}

scokmen / jpipe / 23205294858

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous