libsxmp/include/sntl/connection.h

/* 
 * File:   connection.h
 * Author: vdo
 *
 * Created on September 24, 2014, 2:36 AM
 */

#ifndef __ESXC_CONNECTION_H_
#define	__ESXC_CONNECTION_H_

#include <stdint.h>
#include <time.h>
#include <sys/types.h>
#include <pthread.h>

#include <openssl/ssl.h>

#include <tdata/usrtc.h>
#include <tdata/idx_allocator.h>
#include <sexpr/sexp.h>
#include <sexpr/faststack.h>

#include <sntl/pth_queue.h>

/* sexp helpers */
#define SEXP_IS_LIST(sx) \
        ((sx)->ty == SEXP_LIST) ? 1 : 0

#define SEXP_IS_TYPE(sx,type) \
        ((sx)->ty == SEXP_VALUE && (sx)->aty == (type)) ? 1 : 0

#define SEXP_ITERATE_LIST(lst, iter, ind) \
        for((ind) = 0, (iter) = (lst)->list; (ind) < sexp_list_length(lst); \
            (ind)++, (iter) = (iter)->next)

int sexp_list_cdr(sexp_t *expr, sexp_t **sx);
int sexp_list_car(sexp_t *expr, sexp_t **sx);

#define VERIFY_DEPTH  1  /* FIXME: */

#define MAX_CONNECTIONS  32768
#define MAX_CHANNELS     4096
#define MAX_RPC_LIST     2048
#define MAX_MULTI        12
#define MAX_PENDINGMSG   16384
#define MAX_MSGINDEX     ((MAX_PENDINGMSG) * (MAX_MULTI))

typedef struct __perm_context_type {
  char *login;
  char *passwd;
  ulong_t certid;
  ulong_t uid;
  ulong_t gid;
  ulong_t *gids;
  int n_gids;
  int p_attr;
  void *priv;
} perm_ctx_t;

#define CXCONN_MASTER  (1 << 1)
#define CXCONN_SLAVE   (1 << 2)
#define CXCONN_ESTABL  (1 << 3)

typedef struct __connection_t {
  char *uuid; /** < uuid of the connection */
  idx_allocator_t *idx_ch; /** < index allocation for channels */
  usrtc_t *chnl_tree; /** < search tree of all channels */
  usrtc_t *rpc_list; /** < search tree of possible RPC typed lists */
  SSL_CTX *ctx; /** < SSL context */
  SSL *ssl; /** < SSL connection */
  int ssl_data_index; /** < SSL index for the custom data */
  perm_ctx_t *pctx; /** < higher layer authentification context */
  pthread_t cthread; /** < thread for listening the connection socket */
  pthread_mutex_t oplock; /** < mutex used to sync operations on connection */
  pthread_rwlock_t chnl_lock; /** < rwlock used to sync ops with channels */
  int flags; /** < flags of the connection */
  usrtc_node_t csnode; /** < node to store the connection within list */
} conn_t;

struct __connection_rpc_list_type;
struct __message_t;

#define ESXCHAN_PENDING  (1 << 1)
#define ESXCHAN_CLOSURE  (1 << 2)

typedef struct __channel_t {
  ulong_t cid; /** < ID of the channel */
  char *uuid; /** < UUID of the channel, used in advanced implementation
                * of the complex distributed systems */
  conn_t *connection; /** < pointer to the parent connection */
  idx_allocator_t *idx_msg; /** < index allocation for messages */
  usrtc_t *msgs_tree; /** < search tree of the existing messages */
  struct __message_t *sysmsg; /** < system message used to operate with channel */
  struct __connection_rpc_list_type *rpc_list; /** < rpc functions list */
  pthread_mutex_t oplock; /** < operation ops lock */
  pthread_rwlock_t msglock; /** < rwlock used to operate with messages */
  usrtc_node_t node; /** < node for connection search tree */
  int use_count; /** < use count */
  int flags; /** < flags of the channel */
} chnl_t;

typedef struct __sexp_payload_t {
  char *cstr;
  sexp_t *sx;
} sxpayload_t;

#define ESX_SYSMSG_SIZE  512

#define ESXMSG_SYS      (1 << 1)
#define ESXMSG_USR      (1 << 2)
#define ESXMSG_PENDING  (1 << 3)
#define ESXMSG_NOWAY    (1 << 4)

typedef struct __message_t {
  chnl_t *pch; /** < channel of the message(if applicable) */
  ulong_t mid; /** < unique ID within connection context */
  char *uuid; /** < UUID of the message, used for special messages */
  usrtc_node_t chnl_node; /** < node for channel search tree */
  usrtc_node_t poll_node; /** < node for the poll of the messages */
  usrtc_node_t pendingq_node; /** < node for the pending queue */
  pthread_mutex_t wait; /** < special wait mutex, used for sync */
  void *payload; /** < payload */
  int opcode; /** < opcode for system and pulse messages */
  int flags; /** < flags of the message (type, state etc ...)*/
  int use_count; /** < use count */
} sxmsg_t;

typedef struct __connection_rpc_entry_type {
  char *name;
  int (*rpcf)(void *, sexp_t *);
  usrtc_node_t node;
} cx_rpc_t;

typedef struct __connection_rpc_list_type {
  usrtc_t *rpc_tree; /** < search tree for the rpc lookup */
  char *opt_version; /** < reserved for future implementations */
} cx_rpc_list_t;

typedef struct __connections_subsys_type {
  int ex_ssldata_index; /** < index used to work with additional data
                         * provided to the special call during SSL handshake */
  usrtc_t *connections;
  pth_queue_t *ioq; /** < general messages queue */
  pth_queue_t *ioqueue; /** < system messages queue */
  /* system threads */
  pthread_t iog_thread; /** < general io queue */
  pthread_t ios_thread; /** < system io queue */
  pthread_rwlock_t rwlock;
  char *rootca, *certpem, *certkey; /* path name to the certificates */
  cx_rpc_list_t *system_rpc;
  /* special functions pointers */
  int (*validate_sslpem)(conn_t *); /** < this function used to validate SSL certificate while SSL handshake */
  int (*secure_check)(conn_t *); /** < this function authorize user to login,
                                  * and also should check SSL cert and user, and already made sessions */
  usrtc_t* (*get_rpc_typed_list_tree)(conn_t *); /** < this function is used to set RPC list of the functions */
  int (*set_typed_list_callback)(conn_t *, int, char *); /** < this function is a callback
                                                          * during setting up a typed channel */
  void *priv;
} conn_sys_t;

typedef struct __rpc_typed_list_type {
  int type_id;
  char *description;
  cx_rpc_list_t *rpc_list;
  usrtc_node_t lnode;
} rpc_typed_list_t;

extern conn_sys_t *conn_sys;

/* General API */
/* subsystem */
int connections_subsystem_init(void);

int connections_subsystem_setsslserts(const char *rootca, const char *certpem,
        const char *certkey);

int connections_subsystem_setrpclist_function(usrtc_t* (*get_rpc_typed_list_tree)(conn_t *));

#define connections_subsystem_set_securecheck(c, fuu)    (c)->secure_check = fuu
#define connections_subsystem_set_sslvalidator(c, fuu)   (c)->validate_sslpem = fuu
#define connections_subsystem_set_rpctlist_call(c, fuu)  (c)->set_typed_list_callback = fuu

/* connection */
int connection_initiate (conn_t *co, const char *host, int port, 
        const char *SSL_cert, perm_ctx_t *pctx);

int connection_create(conn_t *co, int sck);

int connection_close(conn_t *co);

int connection_reinit(conn_t *co);

/* channels */
int channel_open(conn_t *co, chnl_t **ch, int type);

int channel_close(conn_t *co);

/* message passing */
int msg_send(chnl_t *ch, sexp_t *sx, sxmsg_t **msg);

int msg_send_timed(chnl_t *ch, sexp_t *sx, sxmsg_t **msg, struct timespec *tio);

int msg_return(sxmsg_t *msg, int opcode);

int msg_reply(sxmsg_t *msg, sexp_t *sx);

int msg_reply_timed(sxmsg_t *msg, sexp_t *sx, struct timespec *tio);

int msg_send_pulse(chnl_t *ch, sexp_t *sx);

int msg_send_pulse_timed(chnl_t *ch, sexp_t *sx, struct timespec *tio);

int msg_send_pulse_nowait(chnl_t *ch, sexp_t *sx);

#endif	/* __ESXC_CONNECTION_H_ */