123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551 |
- #include <stdlib.h>
- #include <stdio.h>
- #include <string.h>
- #include <assert.h>
- #include <unistd.h>
- #include <errno.h>
- #include <sys/types.h>
- #include <sys/stat.h>
- #include <fcntl.h>
- #include <sys/socket.h>
- #include <sys/un.h>
- #include <syslog.h>
- #include <faux/str.h>
- #include <klish/ktp_session.h>
- typedef struct cb_s {
- void *fn;
- void *udata;
- } cb_t;
- struct ktp_session_s {
- ktp_session_state_e state;
- faux_async_t *async;
- faux_hdr_t *hdr; // Service var: engine will receive header and then msg
- bool_t done;
- faux_eloop_t *eloop; // External eloop object
- cb_t cb[KTP_SESSION_CB_MAX];
- faux_error_t *error; // Internal
- bool_t request_done;
- int cmd_retcode; // Internal
- bool_t cmd_retcode_available;
- ktp_status_e cmd_features;
- bool_t cmd_features_available;
- bool_t stop_on_answer; // Stop the loop when answer is received (for non-interactive mode)
- };
- static bool_t server_ev(faux_eloop_t *eloop, faux_eloop_type_e type,
- void *associated_data, void *user_data);
- static bool_t ktp_session_read_cb(faux_async_t *async,
- faux_buf_t *buf, size_t len, void *user_data);
- ktp_session_t *ktp_session_new(int sock, faux_eloop_t *eloop)
- {
- ktp_session_t *ktp = NULL;
- if (sock < 0)
- return NULL;
- if (!eloop)
- return NULL;
- ktp = faux_zmalloc(sizeof(*ktp));
- assert(ktp);
- if (!ktp)
- return NULL;
- // Init
- ktp->state = KTP_SESSION_STATE_IDLE;
- ktp->done = BOOL_FALSE;
- ktp->eloop = eloop;
- ktp->stop_on_answer = BOOL_TRUE; // Non-interactive by default
- ktp->error = NULL;
- ktp->cmd_retcode = -1;
- ktp->cmd_retcode_available = BOOL_FALSE;
- ktp->request_done = BOOL_FALSE;
- ktp->cmd_features = KTP_STATUS_NONE;
- ktp->cmd_features_available = BOOL_FALSE;
- // Async object
- ktp->async = faux_async_new(sock);
- assert(ktp->async);
- // Receive message header first
- faux_async_set_read_limits(ktp->async,
- sizeof(faux_hdr_t), sizeof(faux_hdr_t));
- faux_async_set_read_cb(ktp->async, ktp_session_read_cb, ktp);
- ktp->hdr = NULL;
- faux_async_set_stall_cb(ktp->async, ktp_stall_cb, ktp->eloop);
- // Event loop handlers
- faux_eloop_add_fd(ktp->eloop, ktp_session_fd(ktp), POLLIN,
- server_ev, ktp);
- // Callbacks
- // Callbacks ktp->cb are zeroed by faux_zmalloc()
- return ktp;
- }
- void ktp_session_free(ktp_session_t *ktp)
- {
- if (!ktp)
- return;
- // Remove socket from eloop but don't free eloop because it's external
- faux_eloop_del_fd(ktp->eloop, ktp_session_fd(ktp));
- faux_free(ktp->hdr);
- close(ktp_session_fd(ktp));
- faux_async_free(ktp->async);
- faux_free(ktp);
- }
- faux_eloop_t *ktp_session_eloop(const ktp_session_t *ktp)
- {
- assert(ktp);
- if (!ktp)
- return NULL;
- return ktp->eloop;
- }
- bool_t ktp_session_done(const ktp_session_t *ktp)
- {
- assert(ktp);
- if (!ktp)
- return BOOL_TRUE; // Done flag
- return ktp->done;
- }
- bool_t ktp_session_set_done(ktp_session_t *ktp, bool_t done)
- {
- assert(ktp);
- if (!ktp)
- return BOOL_FALSE;
- ktp->done = done;
- return BOOL_TRUE;
- }
- bool_t ktp_session_stop_on_answer(const ktp_session_t *ktp)
- {
- assert(ktp);
- if (!ktp)
- return BOOL_TRUE; // Default
- return ktp->stop_on_answer;
- }
- bool_t ktp_session_set_stop_on_answer(ktp_session_t *ktp, bool_t stop_on_answer)
- {
- assert(ktp);
- if (!ktp)
- return BOOL_FALSE;
- ktp->stop_on_answer = stop_on_answer;
- return BOOL_TRUE;
- }
- ktp_session_state_e ktp_session_state(const ktp_session_t *ktp)
- {
- assert(ktp);
- if (!ktp)
- return KTP_SESSION_STATE_ERROR;
- return ktp->state;
- }
- ktp_status_e ktp_session_cmd_features(const ktp_session_t *ktp)
- {
- assert(ktp);
- if (!ktp)
- return KTP_STATUS_NONE;
- return ktp->cmd_features;
- }
- faux_error_t *ktp_session_error(const ktp_session_t *ktp)
- {
- assert(ktp);
- if (!ktp)
- return BOOL_FALSE;
- return ktp->error;
- }
- bool_t ktp_session_set_cb(ktp_session_t *ktp, ktp_session_cb_e cb_id,
- void *fn, void *udata)
- {
- assert(ktp);
- if (!ktp)
- return BOOL_FALSE;
- if (cb_id >= KTP_SESSION_CB_MAX)
- return BOOL_FALSE;
- ktp->cb[cb_id].fn = fn;
- ktp->cb[cb_id].udata = udata;
- return BOOL_TRUE;
- }
- bool_t ktp_session_connected(ktp_session_t *ktp)
- {
- assert(ktp);
- if (!ktp)
- return BOOL_FALSE;
- if (KTP_SESSION_STATE_DISCONNECTED == ktp->state)
- return BOOL_FALSE;
- return BOOL_TRUE;
- }
- int ktp_session_fd(const ktp_session_t *ktp)
- {
- assert(ktp);
- if (!ktp)
- return BOOL_FALSE;
- return faux_async_fd(ktp->async);
- }
- static bool_t server_ev(faux_eloop_t *eloop, faux_eloop_type_e type,
- void *associated_data, void *user_data)
- {
- faux_eloop_info_fd_t *info = (faux_eloop_info_fd_t *)associated_data;
- ktp_session_t *ktp = (ktp_session_t *)user_data;
- assert(ktp);
- // Write data
- if (info->revents & POLLOUT) {
- faux_eloop_exclude_fd_event(eloop, info->fd, POLLOUT);
- if (faux_async_out(ktp->async) < 0) {
- // Someting went wrong
- faux_eloop_del_fd(eloop, info->fd);
- syslog(LOG_ERR, "Problem with async output");
- return BOOL_FALSE; // Stop event loop
- }
- }
- // Read data
- if (info->revents & POLLIN) {
- if (faux_async_in(ktp->async) < 0) {
- // Someting went wrong
- faux_eloop_del_fd(eloop, info->fd);
- syslog(LOG_ERR, "Problem with async input");
- return BOOL_FALSE; // Stop event loop
- }
- }
- // EOF
- if (info->revents & POLLHUP) {
- faux_eloop_del_fd(eloop, info->fd);
- syslog(LOG_DEBUG, "Close connection %d", info->fd);
- return BOOL_FALSE; // Stop event loop
- }
- type = type; // Happy compiler
- if (ktp->request_done && ktp->stop_on_answer)
- return BOOL_FALSE; // Stop event loop on receiving answer
- if (ktp->done)
- return BOOL_FALSE; // Stop event loop on done flag (exit)
- return BOOL_TRUE;
- }
- static bool_t ktp_session_process_stdout(ktp_session_t *ktp, const faux_msg_t *msg)
- {
- char *line = NULL;
- unsigned int len = 0;
- assert(ktp);
- assert(msg);
- if (!ktp->cb[KTP_SESSION_CB_STDOUT].fn)
- return BOOL_TRUE; // Just ignore stdout. It's not a bug
- if (!faux_msg_get_param_by_type(msg, KTP_PARAM_LINE, (void **)&line, &len))
- return BOOL_TRUE; // It's strange but not a bug
- return ((ktp_session_stdout_cb_fn)ktp->cb[KTP_SESSION_CB_STDOUT].fn)(
- ktp, line, len, ktp->cb[KTP_SESSION_CB_STDOUT].udata);
- }
- static bool_t ktp_session_process_stderr(ktp_session_t *ktp, const faux_msg_t *msg)
- {
- char *line = NULL;
- unsigned int len = 0;
- assert(ktp);
- assert(msg);
- if (!ktp->cb[KTP_SESSION_CB_STDERR].fn)
- return BOOL_TRUE; // Just ignore message. It's not a bug
- if (!faux_msg_get_param_by_type(msg, KTP_PARAM_LINE,
- (void **)&line, &len))
- return BOOL_TRUE; // It's strange but not a bug
- return ((ktp_session_stdout_cb_fn)ktp->cb[KTP_SESSION_CB_STDERR].fn)(
- ktp, line, len, ktp->cb[KTP_SESSION_CB_STDERR].udata);
- }
- static bool_t ktp_session_process_cmd_ack(ktp_session_t *ktp, const faux_msg_t *msg)
- {
- uint8_t *retcode8bit = NULL;
- ktp_status_e status = KTP_STATUS_NONE;
- char *error_str = NULL;
- assert(ktp);
- assert(msg);
- status = faux_msg_get_status(msg);
- // cmd_ack with flag 'incompleted'
- if (KTP_STATUS_IS_INCOMPLETED(status)) {
- // Only first 'incompleted' cmd ack sets cmd features
- if (!ktp->cmd_features_available) {
- ktp->cmd_features_available = BOOL_TRUE;
- ktp->cmd_features = status & KTP_STATUS_INTERACTIVE;
- }
- // Execute external callback
- if (ktp->cb[KTP_SESSION_CB_CMD_ACK_INCOMPLETED].fn)
- ((ktp_session_event_cb_fn)
- ktp->cb[KTP_SESSION_CB_CMD_ACK_INCOMPLETED].fn)(
- ktp, msg,
- ktp->cb[KTP_SESSION_CB_CMD_ACK_INCOMPLETED].udata);
- return BOOL_TRUE;
- }
- if (faux_msg_get_param_by_type(msg, KTP_PARAM_RETCODE,
- (void **)&retcode8bit, NULL))
- ktp->cmd_retcode = (int)(*retcode8bit);
- error_str = faux_msg_get_str_param_by_type(msg, KTP_PARAM_ERROR);
- if (error_str) {
- faux_error_add(ktp->error, error_str);
- faux_str_free(error_str);
- }
- ktp->cmd_retcode_available = BOOL_TRUE; // Answer from server was received
- ktp->request_done = BOOL_TRUE;
- ktp->state = KTP_SESSION_STATE_IDLE;
- // Get exit flag from message
- if (KTP_STATUS_IS_EXIT(faux_msg_get_status(msg)))
- ktp->done = BOOL_TRUE;
- // Execute external callback
- if (ktp->cb[KTP_SESSION_CB_CMD_ACK].fn)
- ((ktp_session_event_cb_fn)
- ktp->cb[KTP_SESSION_CB_CMD_ACK].fn)(
- ktp, msg,
- ktp->cb[KTP_SESSION_CB_CMD_ACK].udata);
- return BOOL_TRUE;
- }
- /*
- static bool_t ktp_session_process_exit(ktp_session_t *ktp, const faux_msg_t *msg)
- {
- assert(ktp);
- assert(msg);
- ktp_session_set_done(ktp, BOOL_TRUE);
- // Execute external callback
- if (ktp->cb[KTP_SESSION_CB_EXIT].fn)
- ((ktp_session_event_cb_fn)
- ktp->cb[KTP_SESSION_CB_EXIT].fn)(
- ktp, msg,
- ktp->cb[KTP_SESSION_CB_EXIT].udata);
- return BOOL_TRUE;
- }
- */
- static bool_t ktp_session_dispatch(ktp_session_t *ktp, faux_msg_t *msg)
- {
- uint16_t cmd = 0;
- bool_t rc = BOOL_TRUE;
- assert(ktp);
- if (!ktp)
- return BOOL_FALSE;
- assert(msg);
- if (!msg)
- return BOOL_FALSE;
- cmd = faux_msg_get_cmd(msg);
- switch (cmd) {
- case KTP_CMD_ACK:
- if (ktp->state != KTP_SESSION_STATE_WAIT_FOR_CMD) {
- syslog(LOG_WARNING, "Unexpected KTP_CMD_ACK was received\n");
- break;
- }
- rc = ktp_session_process_cmd_ack(ktp, msg);
- break;
- case KTP_STDOUT:
- if (ktp->state != KTP_SESSION_STATE_WAIT_FOR_CMD) {
- syslog(LOG_WARNING, "Unexpected KTP_STDOUT was received\n");
- break;
- }
- rc = ktp_session_process_stdout(ktp, msg);
- break;
- case KTP_STDERR:
- if (ktp->state != KTP_SESSION_STATE_WAIT_FOR_CMD) {
- syslog(LOG_WARNING, "Unexpected KTP_STDERR was received\n");
- break;
- }
- rc = ktp_session_process_stderr(ktp, msg);
- break;
- default:
- syslog(LOG_WARNING, "Unsupported command: 0x%04u\n", cmd); // Ignore
- break;
- }
- return rc;
- }
- static bool_t ktp_session_read_cb(faux_async_t *async,
- faux_buf_t *buf, size_t len, void *user_data)
- {
- ktp_session_t *ktp = (ktp_session_t *)user_data;
- faux_msg_t *completed_msg = NULL;
- char *data = NULL;
- assert(async);
- assert(buf);
- assert(ktp);
- // Linearize buffer
- data = malloc(len);
- faux_buf_read(buf, data, len);
- // Receive header
- if (!ktp->hdr) {
- size_t whole_len = 0;
- size_t msg_wo_hdr = 0;
- ktp->hdr = (faux_hdr_t *)data;
- // Check for broken header
- if (!ktp_check_header(ktp->hdr)) {
- faux_free(ktp->hdr);
- ktp->hdr = NULL;
- return BOOL_FALSE;
- }
- whole_len = faux_hdr_len(ktp->hdr);
- // msg_wo_hdr >= 0 because ktp_check_header() validates whole_len
- msg_wo_hdr = whole_len - sizeof(faux_hdr_t);
- // Plan to receive message body
- if (msg_wo_hdr > 0) {
- faux_async_set_read_limits(async,
- msg_wo_hdr, msg_wo_hdr);
- return BOOL_TRUE;
- }
- // Here message is completed (msg body has zero length)
- completed_msg = faux_msg_deserialize_parts(ktp->hdr, NULL, 0);
- // Receive message body
- } else {
- completed_msg = faux_msg_deserialize_parts(ktp->hdr, data, len);
- faux_free(data);
- }
- // Plan to receive msg header
- faux_async_set_read_limits(ktp->async,
- sizeof(faux_hdr_t), sizeof(faux_hdr_t));
- faux_free(ktp->hdr);
- ktp->hdr = NULL; // Ready to recv new header
- // Here message is completed
- #ifdef DEBUG
- // faux_msg_debug(completed_msg);
- #endif
- ktp_session_dispatch(ktp, completed_msg);
- faux_msg_free(completed_msg);
- return BOOL_TRUE;
- }
- static bool_t ktp_session_req(ktp_session_t *ktp, ktp_cmd_e cmd,
- const char *line, faux_error_t *error, bool_t dry_run)
- {
- faux_msg_t *req = NULL;
- ktp_status_e status = KTP_STATUS_NONE;
- assert(ktp);
- if (!ktp)
- return BOOL_FALSE;
- if (ktp->state != KTP_SESSION_STATE_IDLE) {
- faux_error_sprintf(error,
- "Can't create request. Session state is not suitable");
- return BOOL_FALSE;
- }
- // Set dry-run flag
- if (dry_run)
- status |= KTP_STATUS_DRY_RUN;
- req = ktp_msg_preform(cmd, status);
- faux_msg_add_param(req, KTP_PARAM_LINE, line, strlen(line));
- faux_msg_send_async(req, ktp->async);
- faux_msg_free(req);
- // Prepare for loop
- ktp->error = error;
- ktp->cmd_retcode = -1;
- ktp->cmd_retcode_available = BOOL_FALSE;
- ktp->request_done = BOOL_FALSE;
- ktp->cmd_features = KTP_STATUS_NONE;
- ktp->cmd_features_available = BOOL_FALSE;
- return BOOL_TRUE;
- }
- bool_t ktp_session_cmd(ktp_session_t *ktp, const char *line,
- faux_error_t *error, bool_t dry_run)
- {
- if (!ktp_session_req(ktp, KTP_CMD, line, error, dry_run))
- return BOOL_FALSE;
- ktp->state = KTP_SESSION_STATE_WAIT_FOR_CMD;
- return BOOL_TRUE;
- }
- bool_t ktp_session_retcode(ktp_session_t *ktp, int *retcode)
- {
- if (!ktp)
- return BOOL_FALSE;
- if (ktp->cmd_retcode_available && retcode)
- *retcode = ktp->cmd_retcode;
- return ktp->cmd_retcode_available; // Sign of server answer
- }
|