/* * Copyright (C) NGINX, Inc. */ #ifndef _NXT_UNIT_H_INCLUDED_ #define _NXT_UNIT_H_INCLUDED_ #include #include #include #include #include "nxt_version.h" #include "nxt_unit_typedefs.h" enum { NXT_UNIT_OK = 0, NXT_UNIT_ERROR = 1, }; enum { NXT_UNIT_LOG_ALERT = 0, NXT_UNIT_LOG_ERR = 1, NXT_UNIT_LOG_WARN = 2, NXT_UNIT_LOG_NOTICE = 3, NXT_UNIT_LOG_INFO = 4, NXT_UNIT_LOG_DEBUG = 5, }; #define NXT_UNIT_INIT_ENV "NXT_UNIT_INIT" /* * Mostly opaque structure with library state. * * Only user defined 'data' pointer exposed here. The rest is unit * implementation specific and hidden. */ struct nxt_unit_s { void *data; /* User defined data. */ }; /* * Thread context. * * First (main) context is provided 'for free'. To receive and process * requests in other thread, one need to allocate context and use it * further in this thread. */ struct nxt_unit_ctx_s { void *data; /* User context-specific data. */ nxt_unit_t *unit; }; /* * Unit port identification structure. * * Each port can be uniquely identified by listen process id (pid) and port id. * This identification is required to refer the port from different process. */ struct nxt_unit_port_id_s { pid_t pid; uint32_t hash; uint16_t id; }; /* * unit provides port storage which is able to store and find the following * data structures. */ struct nxt_unit_port_s { nxt_unit_port_id_t id; int in_fd; int out_fd; void *data; }; struct nxt_unit_buf_s { char *start; char *free; char *end; }; struct nxt_unit_request_info_s { nxt_unit_t *unit; nxt_unit_ctx_t *ctx; nxt_unit_port_id_t request_port; nxt_unit_port_id_t response_port; nxt_unit_request_t *request; nxt_unit_buf_t *request_buf; nxt_unit_response_t *response; nxt_unit_buf_t *response_buf; uint32_t response_max_fields; nxt_unit_buf_t *content_buf; uint64_t content_length; int content_fd; void *data; }; /* * Set of application-specific callbacks. Application may leave all optional * callbacks as NULL. */ struct nxt_unit_callbacks_s { /* * Process request. Unlike all other callback, this callback * need to be defined by application. */ void (*request_handler)(nxt_unit_request_info_t *req); /* Process websocket frame. */ void (*websocket_handler)(nxt_unit_websocket_frame_t *ws); /* Connection closed. */ void (*close_handler)(nxt_unit_request_info_t *req); /* Add new Unit port to communicate with process pid. Optional. */ int (*add_port)(nxt_unit_ctx_t *, nxt_unit_port_t *port); /* Remove previously added port. Optional. */ void (*remove_port)(nxt_unit_t *, nxt_unit_port_t *port); /* Remove all data associated with process pid including ports. Optional. */ void (*remove_pid)(nxt_unit_t *, pid_t pid); /* Gracefully quit the application. Optional. */ void (*quit)(nxt_unit_ctx_t *); /* Shared memory release acknowledgement. Optional. */ void (*shm_ack_handler)(nxt_unit_ctx_t *); /* Send data and control to process pid using port id. Optional. */ ssize_t (*port_send)(nxt_unit_ctx_t *, nxt_unit_port_id_t *port_id, const void *buf, size_t buf_size, const void *oob, size_t oob_size); /* Receive data on port id. Optional. */ ssize_t (*port_recv)(nxt_unit_ctx_t *, nxt_unit_port_id_t *port_id, void *buf, size_t buf_size, void *oob, size_t oob_size); }; struct nxt_unit_init_s { void *data; /* Opaque pointer to user-defined data. */ void *ctx_data; /* Opaque pointer to user-defined data. */ int max_pending_requests; uint32_t request_data_size; uint32_t shm_limit; nxt_unit_callbacks_t callbacks; nxt_unit_port_t ready_port; uint32_t ready_stream; nxt_unit_port_t router_port; nxt_unit_port_t read_port; int log_fd; }; typedef ssize_t (*nxt_unit_read_func_t)(nxt_unit_read_info_t *read_info, void *dst, size_t size); struct nxt_unit_read_info_s { nxt_unit_read_func_t read; int eof; uint32_t buf_size; void *data; }; /* * Initialize Unit application library with necessary callbacks and * ready/reply port parameters, send 'READY' response to master. */ nxt_unit_ctx_t *nxt_unit_init(nxt_unit_init_t *); /* * Process received message, invoke configured callbacks. * * If application implements it's own event loop, each datagram received * from port socket should be initially processed by unit. This function * may invoke other application-defined callback for message processing. */ int nxt_unit_process_msg(nxt_unit_ctx_t *, nxt_unit_port_id_t *port_id, void *buf, size_t buf_size, void *oob, size_t oob_size); /* * Main function useful in case when application does not have it's own * event loop. nxt_unit_run() starts infinite message wait and process loop. * * for (;;) { * app_lib->port_recv(...); * nxt_unit_process_msg(...); * } * * The normally function returns when QUIT message received from Unit. */ int nxt_unit_run(nxt_unit_ctx_t *); int nxt_unit_run_once(nxt_unit_ctx_t *ctx); /* Destroy application library object. */ void nxt_unit_done(nxt_unit_ctx_t *); /* * Allocate and initialize new execution context with new listen port to * process requests in other thread. */ nxt_unit_ctx_t *nxt_unit_ctx_alloc(nxt_unit_ctx_t *, void *); /* Initialize port_id, calculate hash. */ void nxt_unit_port_id_init(nxt_unit_port_id_t *port_id, pid_t pid, uint16_t id); /* Default 'port_recv' implementation. */ ssize_t nxt_unit_port_recv(nxt_unit_ctx_t *, int fd, void *buf, size_t buf_size, void *oob, size_t oob_size); /* Calculates hash for given field name. */ uint16_t nxt_unit_field_hash(const char* name, size_t name_length); /* Split host for server name and port. */ void nxt_unit_split_host(char *host_start, uint32_t host_length, char **name, uint32_t *name_length, char **port, uint32_t *port_length); /* Group duplicate fields for easy enumeration. */ void nxt_unit_request_group_dup_fields(nxt_unit_request_info_t *req); /* * Allocate response structure capable to store limited numer of fields. * The structure may be accessed directly via req->response pointer or * filled step-by-step using functions add_field and add_content. */ int nxt_unit_response_init(nxt_unit_request_info_t *req, uint16_t status, uint32_t max_fields_count, uint32_t max_fields_size); int nxt_unit_response_realloc(nxt_unit_request_info_t *req, uint32_t max_fields_count, uint32_t max_fields_size); int nxt_unit_response_is_init(nxt_unit_request_info_t *req); int nxt_unit_response_add_field(nxt_unit_request_info_t *req, const char* name, uint8_t name_length, const char* value, uint32_t value_length); int nxt_unit_response_add_content(nxt_unit_request_info_t *req, const void* src, uint32_t size); /* * Send prepared response to Unit server. Response structure destroyed during * this call. */ int nxt_unit_response_send(nxt_unit_request_info_t *req); int nxt_unit_response_is_sent(nxt_unit_request_info_t *req); nxt_unit_buf_t *nxt_unit_response_buf_alloc(nxt_unit_request_info_t *req, uint32_t size); int nxt_unit_request_is_websocket_handshake(nxt_unit_request_info_t *req); int nxt_unit_response_upgrade(nxt_unit_request_info_t *req); int nxt_unit_response_is_websocket(nxt_unit_request_info_t *req); nxt_unit_request_info_t *nxt_unit_get_request_info_from_data(void *data); int nxt_unit_buf_send(nxt_unit_buf_t *buf); void nxt_unit_buf_free(nxt_unit_buf_t *buf); nxt_unit_buf_t *nxt_unit_buf_next(nxt_unit_buf_t *buf); uint32_t nxt_unit_buf_max(void); uint32_t nxt_unit_buf_min(void); int nxt_unit_response_write(nxt_unit_request_info_t *req, const void *start, size_t size); ssize_t nxt_unit_response_write_nb(nxt_unit_request_info_t *req, const void *start, size_t size, size_t min_size); int nxt_unit_response_write_cb(nxt_unit_request_info_t *req, nxt_unit_read_info_t *read_info); ssize_t nxt_unit_request_read(nxt_unit_request_info_t *req, void *dst, size_t size); ssize_t nxt_unit_request_readline_size(nxt_unit_request_info_t *req, size_t max_size); void nxt_unit_request_done(nxt_unit_request_info_t *req, int rc); int nxt_unit_websocket_send(nxt_unit_request_info_t *req, uint8_t opcode, uint8_t last, const void *start, size_t size); int nxt_unit_websocket_sendv(nxt_unit_request_info_t *req, uint8_t opcode, uint8_t last, const struct iovec *iov, int iovcnt); ssize_t nxt_unit_websocket_read(nxt_unit_websocket_frame_t *ws, void *dst, size_t size); int nxt_unit_websocket_retain(nxt_unit_websocket_frame_t *ws); void nxt_unit_websocket_done(nxt_unit_websocket_frame_t *ws); #if defined __has_attribute #if __has_attribute(format) #define NXT_ATTR_FORMAT __attribute__((format(printf, 3, 4))) #endif #endif #if !defined(NXT_ATTR_FORMAT) #define NXT_ATTR_FORMAT #endif void nxt_unit_log(nxt_unit_ctx_t *ctx, int level, const char* fmt, ...) NXT_ATTR_FORMAT; void nxt_unit_req_log(nxt_unit_request_info_t *req, int level, const char* fmt, ...) NXT_ATTR_FORMAT; #if (NXT_DEBUG) #define nxt_unit_debug(ctx, fmt, ARGS...) \ nxt_unit_log(ctx, NXT_UNIT_LOG_DEBUG, fmt, ##ARGS) #define nxt_unit_req_debug(req, fmt, ARGS...) \ nxt_unit_req_log(req, NXT_UNIT_LOG_DEBUG, fmt, ##ARGS) #else #define nxt_unit_debug(ctx, fmt, ARGS...) #define nxt_unit_req_debug(req, fmt, ARGS...) #endif #define nxt_unit_warn(ctx, fmt, ARGS...) \ nxt_unit_log(ctx, NXT_UNIT_LOG_WARN, fmt, ##ARGS) #define nxt_unit_req_warn(req, fmt, ARGS...) \ nxt_unit_req_log(req, NXT_UNIT_LOG_WARN, fmt, ##ARGS) #define nxt_unit_error(ctx, fmt, ARGS...) \ nxt_unit_log(ctx, NXT_UNIT_LOG_ERR, fmt, ##ARGS) #define nxt_unit_req_error(req, fmt, ARGS...) \ nxt_unit_req_log(req, NXT_UNIT_LOG_ERR, fmt, ##ARGS) #define nxt_unit_alert(ctx, fmt, ARGS...) \ nxt_unit_log(ctx, NXT_UNIT_LOG_ALERT, fmt, ##ARGS) #define nxt_unit_req_alert(req, fmt, ARGS...) \ nxt_unit_req_log(req, NXT_UNIT_LOG_ALERT, fmt, ##ARGS) #endif /* _NXT_UNIT_H_INCLUDED_ */