From: Miloš Mulač Date: Tue, 16 Nov 2004 13:11:00 +0000 (+0000) Subject: - QuerySequenceCode() X-Git-Tag: glite-lb-client-interface_R_0_3_0~92 X-Git-Url: http://scientific.zcu.cz/git/?a=commitdiff_plain;h=3ba5457e7b7460404e2cde6c8a9cb6de7f03648d;p=jra1mw.git - QuerySequenceCode() - plain communication with proxy not tested yet --- diff --git a/org.glite.lb.client-interface/interface/consumer.h b/org.glite.lb.client-interface/interface/consumer.h index ccd1b37..74b12a3 100644 --- a/org.glite.lb.client-interface/interface/consumer.h +++ b/org.glite.lb.client-interface/interface/consumer.h @@ -268,6 +268,21 @@ int edg_wll_QueryListener( uint16_t * port ); + +/** + * Ask LB Proxy server for sequence number + * \param context IN: context to work with + * \param jobId IN: job to query + * \param code OUT: sequence code + */ + + +int edg_wll_QuerySequenceCode( + edg_wll_Context context, + edg_wlc_JobId jobId, + char ** code +); + /*@}*/ /* @@ -277,6 +292,8 @@ int edg_wll_QueryListener( /** Free edg_wll_QueryRec internals, not the structure itself */ void edg_wll_QueryRecFree(edg_wll_QueryRec *); + + /** * default and maximal query timeout (in seconds) */ diff --git a/org.glite.lb.client/src/connection.c b/org.glite.lb.client/src/connection.c index 54d2f9e..f6b4068 100644 --- a/org.glite.lb.client/src/connection.c +++ b/org.glite.lb.client/src/connection.c @@ -93,7 +93,6 @@ static void ReleaseConnection(edg_wll_Context ctx, char *name, int port) - int edg_wll_close(edg_wll_Context ctx) { edg_wll_ResetError(ctx); @@ -105,6 +104,16 @@ int edg_wll_close(edg_wll_Context ctx) +int edg_wll_close_proxy(edg_wll_Context ctx) +{ + close(ctx->connPlain->sock); + ctx->connPlain->sock = 0; + + return edg_wll_Error(ctx,NULL,NULL); +} + + + int edg_wll_open(edg_wll_Context ctx) { int index; @@ -181,6 +190,55 @@ ok: +int edg_wll_open_proxy(edg_wll_Context ctx) +{ + struct sockaddr_un saddr; + int flags; + + + ctx->connPlain->sock = socket(PF_UNIX, SOCK_STREAM, 0); + if (ctx->connPlain->sock < 0) { + edg_wll_SetError(ctx, errno, "socket() error"); + goto err; + } + + memset(&saddr, 0, sizeof(saddr)); + saddr.sun_family = AF_UNIX; + if (!ctx->p_proxy_filename) { + edg_wll_SetError(ctx, EINVAL, "Proxy socket path not set!"); + goto err; + } + + if (strlen(ctx->p_proxy_filename) > 108) { // UNIX_PATH_MAX (def. in linux/un.h) + // but not defined in sys/un.h + edg_wll_SetError(ctx, EINVAL, "proxy_filename too long!"); + goto err; + } + strcpy(saddr.sun_path, ctx->p_proxy_filename); + + if ((flags = fcntl(ctx->connPlain->sock, F_GETFL, 0)) < 0 || + fcntl(ctx->connPlain->sock, F_SETFL, flags | O_NONBLOCK) < 0) { + edg_wll_SetError(ctx, errno, "fcntl()"); + goto err; + } + + if (connect(ctx->connPlain->sock, (struct sockaddr *)&saddr, sizeof(saddr)) < 0) { + edg_wll_SetError(ctx, errno, "connect()"); + goto err; + } + + return edg_wll_Error(ctx,NULL,NULL); + +err: + /* some error occured; close created connection */ + + edg_wll_close_proxy(ctx); + + return edg_wll_Error(ctx,NULL,NULL); +} + + + /* transform HTTP error code to ours */ int http_check_status( edg_wll_Context ctx, @@ -259,3 +317,37 @@ int edg_wll_http_send_recv( return edg_wll_Error(ctx,NULL,NULL); } + + + +int edg_wll_http_send_recv_proxy( + edg_wll_Context ctx, + char *request, + const char * const *req_head, + char *req_body, + char **response, + char ***resp_head, + char **resp_body) +{ + if (edg_wll_open_proxy(ctx)) return edg_wll_Error(ctx,NULL,NULL); + + switch (edg_wll_http_send_proxy(ctx,request,req_head,req_body)) { + case ENOTCONN: + edg_wll_close_proxy(ctx); + if (edg_wll_open_proxy(ctx) + || edg_wll_http_send_proxy(ctx,request,req_head,req_body)) + return edg_wll_Error(ctx,NULL,NULL); + /* fallthrough */ + case 0: break; + default: return edg_wll_Error(ctx,NULL,NULL); + } + + if (edg_wll_http_recv_proxy(ctx,response,resp_head,resp_body) == ENOTCONN) { + edg_wll_close_proxy(ctx); + (void) (edg_wll_open_proxy(ctx) + || edg_wll_http_send_proxy(ctx,request,req_head,req_body) + || edg_wll_http_recv_proxy(ctx,response,resp_head,resp_body)); + } + + return edg_wll_Error(ctx,NULL,NULL); +} diff --git a/org.glite.lb.client/src/connection.h b/org.glite.lb.client/src/connection.h index 99dd933..7de349d 100644 --- a/org.glite.lb.client/src/connection.h +++ b/org.glite.lb.client/src/connection.h @@ -6,6 +6,11 @@ int edg_wll_close(edg_wll_Context ctx); int edg_wll_open(edg_wll_Context ctx); int edg_wll_http_send_recv(edg_wll_Context, char *, const char * const *, char *, char **, char ***, char **); + +int edg_wll_close_proxy(edg_wll_Context ctx); +int edg_wll_open_proxy(edg_wll_Context ctx); +int edg_wll_http_send_recv_proxy(edg_wll_Context, char *, const char * const *, char *, char **, char ***, char **); + int http_check_status(edg_wll_Context, char *); diff --git a/org.glite.lb.client/src/consumer.c b/org.glite.lb.client/src/consumer.c index 054d9fe..09abdd5 100644 --- a/org.glite.lb.client/src/consumer.c +++ b/org.glite.lb.client/src/consumer.c @@ -365,6 +365,43 @@ int edg_wll_QueryListener(edg_wll_Context ctx, edg_wlc_JobId job, const char *na +int edg_wll_QuerySequenceCode(edg_wll_Context ctx, edg_wlc_JobId jobId, char **code) +{ + int error = 0; + char *response = NULL, + *message = NULL, + *send_mess = NULL; + + + edg_wll_ResetError(ctx); + + if ( edg_wll_QuerySequenceCodeToXML(ctx, jobId, &send_mess) != 0 ) + { + edg_wll_SetError(ctx , (edg_wll_ErrorCode) EINVAL, "Invalid query record."); + goto err; + } + + // ctx->p_tmp_timeout = ctx->p_query_timeout; // not used + + error = edg_wll_http_send_recv_proxy(ctx, "POST /querySequenceCode HTTP/1.1", + request_headers, send_mess, &response, NULL, &message); + if ( error != 0 ) + goto err; + + if (http_check_status(ctx,response)) + goto err; + + edg_wll_ParseQuerySequenceCodeResult(ctx,message,code); + +err: + free(response); + free(message); + free(send_mess); + return edg_wll_Error(ctx,NULL,NULL); + +} + + int set_server_name_and_port(edg_wll_Context ctx, const edg_wll_QueryRec **job_conditions) { int i = 0, j, diff --git a/org.glite.lb.client/src/producer.c b/org.glite.lb.client/src/producer.c index bfe0eb3..673d76b 100644 --- a/org.glite.lb.client/src/producer.c +++ b/org.glite.lb.client/src/producer.c @@ -712,23 +712,34 @@ int edg_wll_SetLoggingJobProxy( int flags) { int err; + char *code_loc; edg_wll_ResetError(context); +/* XXX: add user credentials somewhere - to context? */ + edg_wll_SetParamString(context, EDG_WLL_PARAM_LBPROXY_USER, user); + if (!job) return edg_wll_SetError(context,EINVAL,"jobid is null"); edg_wlc_JobIdFree(context->p_jobid); - if ((err = edg_wlc_JobIdDup(job,&context->p_jobid))) + if ((err = edg_wlc_JobIdDup(job,&context->p_jobid))) { edg_wll_SetError(context,err,"edg_wll_SetLoggingJob(): edg_wlc_JobIdDup() error"); + goto err; + } - else { - if (!edg_wll_SetSequenceCode(context,code,flags)) + /* query LBProxyServer for sequence code if not user-suplied *? + if (!code) { + edg_wll_QuerySequenceCode(context, job, &code_loc); + goto err; + } + else + code_loc = code; + + if (!edg_wll_SetSequenceCode(context,code_loc,flags)) /* XXX: ask proxy for last known sequence code */ - edg_wll_IncSequenceCode(context); - } -/* XXX: add user credentials somewhere - to context? */ - edg_wll_SetParamString(context, EDG_WLL_PARAM_LBPROXY_USER, user); - + edg_wll_IncSequenceCode(context); + +err: return edg_wll_Error(context,NULL,NULL); } diff --git a/org.glite.lb.common/interface/context-int.h b/org.glite.lb.common/interface/context-int.h index 4593e1a..c8125cb 100644 --- a/org.glite.lb.common/interface/context-int.h +++ b/org.glite.lb.common/interface/context-int.h @@ -5,13 +5,12 @@ #include "glite/lb/consumer.h" #include "lb_gss.h" -#include "lb_plain_io.h" +//#include "lb_plain_io.h" #include "authz.h" #ifdef __cplusplus extern "C" { #endif - typedef struct _edg_wll_SeqCode { unsigned int c[EDG_WLL_SOURCE__LAST]; @@ -19,6 +18,16 @@ typedef struct _edg_wll_SeqCode { +/* non-gsi one-element analogy of connPool */ +typedef struct _edg_wll_Connection { + int sock; + char *buf; + size_t bufSize; + size_t bufUse; +} edg_wll_Connection; + + + typedef struct _edg_wll_ConnPool { /* address and port where we are connected to */ char *peerName; @@ -47,6 +56,7 @@ struct _edg_wll_Context { void *mysql; edg_wll_ConnPool *connPool; edg_wll_ConnPool *connPoolNotif; /* hold _one_ connection from notif-interlogger */ + edg_wll_Connection *connPlain; /* holds one plain connection */ int semaphores,semset; edg_wll_QueryRec **job_index; diff --git a/org.glite.lb.common/interface/lb_plain_io.h b/org.glite.lb.common/interface/lb_plain_io.h index c7c8cf1..29c5032 100644 --- a/org.glite.lb.common/interface/lb_plain_io.h +++ b/org.glite.lb.common/interface/lb_plain_io.h @@ -1,18 +1,15 @@ #ifndef __EDG_WORKLOAD_LOGGING_COMMON_LB_PLAIN_IO_H__ #define __EDG_WORKLOAD_LOGGING_COMMON_LB_PLAIN_IO_H__ + +#include "context-int.h" + #ifdef __cplusplus extern "C" { #endif -typedef struct _edg_wll_Connection { - int sock; - char *buffer; - size_t bufsz; - size_t bufuse; -} edg_wll_Connection; - + int edg_wll_plain_accept( int sock, edg_wll_Connection *conn); diff --git a/org.glite.lb.common/interface/xml_conversions.h b/org.glite.lb.common/interface/xml_conversions.h index 246c488..7d289ab 100644 --- a/org.glite.lb.common/interface/xml_conversions.h +++ b/org.glite.lb.common/interface/xml_conversions.h @@ -67,6 +67,9 @@ typedef struct _edg_wll_XML_ctx { edg_wll_NotifId notifId; edg_wll_NotifChangeOp notifChangeOp; time_t notifValidity; + edg_wlc_JobId jobId; + char *source; + char *seqCode; int errCode; int bound; /* marks 2nd value of within operator */ char *errDesc; diff --git a/org.glite.lb.common/interface/xml_parse.h b/org.glite.lb.common/interface/xml_parse.h index e719598..a08d788 100644 --- a/org.glite.lb.common/interface/xml_parse.h +++ b/org.glite.lb.common/interface/xml_parse.h @@ -42,6 +42,8 @@ extern edg_wll_ErrorCode edg_wll_ParseIndexedAttrs(edg_wll_Context ctx, char *me extern edg_wll_ErrorCode edg_wll_ParseNotifResult(edg_wll_Context ctx, char *messageBody, time_t *validity); +extern edg_wll_ErrorCode edg_wll_ParseQuerySequenceCodeResult(edg_wll_Context ctx, char *messageBody, char **seqCode); + extern int edg_wll_QueryEventsRequestToXML(edg_wll_Context ctx, const edg_wll_QueryRec **job_conditions, const edg_wll_QueryRec **event_conditions, char **send_mess); extern int edg_wll_JobQueryRecToXML(edg_wll_Context ctx, edg_wll_QueryRec const * const *conditions, char **send_mess); @@ -58,6 +60,9 @@ extern int edg_wll_IndexedAttrsRequestToXML(edg_wll_Context ctx, char **message) extern int edg_wll_NotifRequestToXML( edg_wll_Context ctx, const char *function, const edg_wll_NotifId notifId, const char *address, edg_wll_NotifChangeOp op, edg_wll_QueryRec const * const *conditions, char **message); +extern int edg_wll_QuerySequenceCodeToXML(edg_wll_Context ctx, edg_wlc_JobId jobId, char **message); + + #ifdef __cplusplus } #endif diff --git a/org.glite.lb.common/src/context.c b/org.glite.lb.common/src/context.c index 2cf1255..43d595e 100644 --- a/org.glite.lb.common/src/context.c +++ b/org.glite.lb.common/src/context.c @@ -32,6 +32,7 @@ int edg_wll_InitContext(edg_wll_Context *ctx) out->connPool = (edg_wll_ConnPool *) calloc(out->poolSize, sizeof(edg_wll_ConnPool)); out->connPoolNotif = (edg_wll_ConnPool *) calloc(1, sizeof(edg_wll_ConnPool)); + out->connPlain = (edg_wll_Connection *) calloc(1, sizeof(edg_wll_Connection)); *ctx = out; return 0; @@ -52,7 +53,7 @@ void edg_wll_FreeContext(edg_wll_Context ctx) if (ctx->connPool[i].peerName) free(ctx->connPool[i].peerName); close(ctx->connPool[i].conn.sock); - if (ctx->connPool[i].conn.buffer) free(ctx->connPool[i].conn.buffer); + if (ctx->connPool[i].conn.buf) free(ctx->connPool[i].conn.buf); edg_wll_gss_close(&ctx->connPool[i].gss,&close_timeout); if (ctx->connPool[i].gsiCred) @@ -61,6 +62,10 @@ void edg_wll_FreeContext(edg_wll_Context ctx) } free(ctx->connPool); } + if (ctx->connPlain) { + if (ctx->connPlain->buf) free(ctx->connPlain->buf); + free(ctx->connPlain); + } if (ctx->notifSock >=0) close(ctx->notifSock); if (ctx->srvName) free(ctx->srvName); if (ctx->peerName) free(ctx->peerName); @@ -215,10 +220,7 @@ static const char* const srcNames[] = { edg_wll_Source edg_wll_StringToSource(const char *name) { int i; -/* XXX: remove - for (i=1; srcNames[i] && strcmp(name,srcNames[i]); i++); - return srcNames[i] ? i : EDG_WLL_SOURCE_NONE; -*/ + for (i=1; ibuffer, (conn)->buffer+(shift), (conn)->bufuse-(shift)); \ - (conn)->bufuse -= (shift); \ + memmove((conn)->buf, (conn)->buf+(shift), (conn)->bufUse-(shift)); \ + (conn)->bufUse -= (shift); \ } int edg_wll_plain_connect( @@ -46,7 +46,7 @@ int edg_wll_plain_accept( /* Do not free the buffer here - just reuse the memmory */ - conn->bufuse = 0; + conn->bufUse = 0; /* if ( (conn->sock = accept(sock, (struct sockaddr *)&a, &alen)) ) return -1; @@ -65,10 +65,10 @@ int edg_wll_plain_read( struct timeval timeout, before, after; - if ( conn->bufsz == 0 ) { - if ( !(conn->buffer = malloc(BUFSIZ)) ) return -1; - conn->bufsz = BUFSIZ; - conn->bufuse = 0; + if ( conn->bufSize == 0 ) { + if ( !(conn->buf = malloc(BUFSIZ)) ) return -1; + conn->bufSize = BUFSIZ; + conn->bufUse = 0; } if ( to ) { @@ -78,7 +78,7 @@ int edg_wll_plain_read( errno = 0; - if ( conn->bufuse > 0 ) goto cleanup; + if ( conn->bufUse > 0 ) goto cleanup; toread = 0; do { @@ -89,24 +89,24 @@ int edg_wll_plain_read( case -1: goto cleanup; break; } - if ( conn->bufuse == conn->bufsz ) { - char *tmp = realloc(conn->buffer, conn->bufsz+BUFSIZ); + if ( conn->bufUse == conn->bufSize ) { + char *tmp = realloc(conn->buf, conn->bufSize+BUFSIZ); if ( !tmp ) return -1; - conn->buffer = tmp; - conn->bufsz += BUFSIZ; + conn->buf = tmp; + conn->bufSize += BUFSIZ; } - toread = conn->bufsz - conn->bufuse; - if ( (ct = read(conn->sock, conn->buffer+conn->bufuse, toread)) < 0 ) { + toread = conn->bufSize - conn->bufUse; + if ( (ct = read(conn->sock, conn->buf+conn->bufUse, toread)) < 0 ) { if ( errno == EINTR ) continue; goto cleanup; } - if ( ct == 0 && conn->bufuse == 0 && errno == 0 ) { + if ( ct == 0 && conn->bufUse == 0 && errno == 0 ) { errno = ENOTCONN; goto cleanup; } - conn->bufuse += ct; + conn->bufUse += ct; } while ( ct == toread ); @@ -120,9 +120,9 @@ cleanup: if ( errno ) return -1; - if ( conn->bufuse > 0 ) { - size_t len = (conn->bufuse < outbufsz) ? conn->bufuse : outbufsz; - memcpy(outbuf, conn->buffer, len); + if ( conn->bufUse > 0 ) { + size_t len = (conn->bufUse < outbufsz) ? conn->bufUse : outbufsz; + memcpy(outbuf, conn->buf, len); outbufsz = len; bufshift(conn, len); return len; @@ -141,9 +141,9 @@ int edg_wll_plain_read_full( size_t total = 0; - if ( conn->bufuse > 0 ) { - size_t len = (conn->bufuse < outbufsz) ? conn->bufuse : outbufsz; - memcpy(outbuf, conn->buffer, len); + if ( conn->bufUse > 0 ) { + size_t len = (conn->bufUse < outbufsz) ? conn->bufUse : outbufsz; + memcpy(outbuf, conn->buf, len); outbufsz = len; bufshift(conn, len); total += len; diff --git a/org.glite.lb.common/src/mini_http.c b/org.glite.lb.common/src/mini_http.c index f3a432c..2ac714e 100644 --- a/org.glite.lb.common/src/mini_http.c +++ b/org.glite.lb.common/src/mini_http.c @@ -14,6 +14,7 @@ #include "mini_http.h" #include "lb_gss.h" +#include "lb_plain_io.h" #include "context-int.h" #define min(x,y) ((x) < (y) ? (x) : (y)) @@ -151,46 +152,46 @@ edg_wll_ErrorCode edg_wll_http_recv_proxy(edg_wll_Context ctx,char **firstOut,ch int len, nhdr = 0,rdmore = 0,clen = 0,blen = 0; #define bshift(shift) {\ - memmove(ctx->connPool[ctx->connToUse].buf,\ - ctx->connPool[ctx->connToUse].buf+(shift),\ - ctx->connPool[ctx->connToUse].bufUse-(shift));\ - ctx->connPool[ctx->connToUse].bufUse -= (shift);\ + memmove(ctx->connPlain->buf,\ + ctx->connPlain->buf+(shift),\ + ctx->connPlain->bufUse-(shift));\ + ctx->connPlain->bufUse -= (shift);\ } edg_wll_ResetError(ctx); - if ( !ctx->connPool[ctx->connToUse].buf ) { - ctx->connPool[ctx->connToUse].bufSize = BUFSIZ; - ctx->connPool[ctx->connToUse].buf = malloc(BUFSIZ); + if ( !ctx->connPlain->buf ) { + ctx->connPlain->bufSize = BUFSIZ; + ctx->connPlain->buf = malloc(BUFSIZ); } do { - len = edg_wll_plain_read(&ctx->connPool[ctx->connToUse].conn, - ctx->connPool[ctx->connToUse].buf+ctx->connPool[ctx->connToUse].bufUse, - ctx->connPool[ctx->connToUse].bufSize-ctx->connPool[ctx->connToUse].bufUse, + len = edg_wll_plain_read(&ctx->connPlain->sock, + ctx->connPlain->buf+ctx->connPlain->bufUse, + ctx->connPlain->bufSize-ctx->connPlain->bufUse, &ctx->p_tmp_timeout); if ( len < 0 ) goto error; - ctx->connPool[ctx->connToUse].bufUse += len; + ctx->connPlain->bufUse += len; rdmore = 0; while (!rdmore && pstat != DONE) switch (pstat) { char *cr; case FIRST: - if ((cr = memchr(ctx->connPool[ctx->connToUse].buf,'\r',ctx->connPool[ctx->connToUse].bufUse)) && - ctx->connPool[ctx->connToUse].bufUse >= cr-ctx->connPool[ctx->connToUse].buf+2 && cr[1] == '\n') + if ((cr = memchr(ctx->connPlain->buf,'\r',ctx->connPlain->bufUse)) && + ctx->connPlain->bufUse >= cr-ctx->connPlain->buf+2 && cr[1] == '\n') { *cr = 0; - first = strdup(ctx->connPool[ctx->connToUse].buf); - bshift(cr-ctx->connPool[ctx->connToUse].buf+2); + first = strdup(ctx->connPlain->buf); + bshift(cr-ctx->connPlain->buf+2); pstat = HEAD; } else rdmore = 1; break; case HEAD: - if ((cr = memchr(ctx->connPool[ctx->connToUse].buf,'\r',ctx->connPool[ctx->connToUse].bufUse)) && - ctx->connPool[ctx->connToUse].bufUse >= cr-ctx->connPool[ctx->connToUse].buf+2 && cr[1] == '\n') + if ((cr = memchr(ctx->connPlain->buf,'\r',ctx->connPlain->bufUse)) && + ctx->connPlain->bufUse >= cr-ctx->connPlain->buf+2 && cr[1] == '\n') { - if (cr == ctx->connPool[ctx->connToUse].buf) { + if (cr == ctx->connPlain->buf) { bshift(2); pstat = clen ? BODY : DONE; if (clen) body = malloc(clen+1); @@ -199,19 +200,19 @@ edg_wll_ErrorCode edg_wll_http_recv_proxy(edg_wll_Context ctx,char **firstOut,ch *cr = 0; hdr = realloc(hdr,(nhdr+2) * sizeof(*hdr)); - hdr[nhdr] = strdup(ctx->connPool[ctx->connToUse].buf); + hdr[nhdr] = strdup(ctx->connPlain->buf); hdr[++nhdr] = NULL; - if (!strncasecmp(ctx->connPool[ctx->connToUse].buf,CONTENT_LENGTH,sizeof(CONTENT_LENGTH)-1)) - clen = atoi(ctx->connPool[ctx->connToUse].buf+sizeof(CONTENT_LENGTH)-1); + if (!strncasecmp(ctx->connPlain->buf,CONTENT_LENGTH,sizeof(CONTENT_LENGTH)-1)) + clen = atoi(ctx->connPlain->buf+sizeof(CONTENT_LENGTH)-1); - bshift(cr-ctx->connPool[ctx->connToUse].buf+2); + bshift(cr-ctx->connPlain->buf+2); } else rdmore = 1; break; case BODY: - if (ctx->connPool[ctx->connToUse].bufUse) { - int m = min(ctx->connPool[ctx->connToUse].bufUse,clen-blen); - memcpy(body+blen,ctx->connPool[ctx->connToUse].buf,m); + if (ctx->connPlain->bufUse) { + int m = min(ctx->connPlain->bufUse,clen-blen); + memcpy(body+blen,ctx->connPlain->buf,m); blen += m; bshift(m); } @@ -330,16 +331,16 @@ edg_wll_ErrorCode edg_wll_http_send_proxy(edg_wll_Context ctx, const char *first edg_wll_ResetError(ctx); - if ( edg_wll_plain_write_full(&ctx->connPool[ctx->connToUse].conn, + if ( edg_wll_plain_write_full(&ctx->connPlain->sock, first, strlen(first), &ctx->p_tmp_timeout) < 0 - || edg_wll_plain_write_full(&ctx->connPool[ctx->connToUse].conn, + || edg_wll_plain_write_full(&ctx->connPlain->sock, "\r\n", 2, &ctx->p_tmp_timeout) < 0 ) return edg_wll_SetError(ctx, errno, "edg_wll_http_send()"); if ( head ) for ( h = head; *h; h++ ) - if ( edg_wll_plain_write_full(&ctx->connPool[ctx->connToUse].conn, + if ( edg_wll_plain_write_full(&ctx->connPlain->sock, *h, strlen(*h), &ctx->p_tmp_timeout) < 0 - || edg_wll_plain_write_full(&ctx->connPool[ctx->connToUse].conn, + || edg_wll_plain_write_full(&ctx->connPlain->sock, "\r\n", 2, &ctx->p_tmp_timeout) < 0 ) return edg_wll_SetError(ctx, errno, "edg_wll_http_send()"); @@ -348,15 +349,15 @@ edg_wll_ErrorCode edg_wll_http_send_proxy(edg_wll_Context ctx, const char *first len = strlen(body); blen = sprintf(buf, CONTENT_LENGTH " %d\r\n",len); - if (edg_wll_plain_write_full(&ctx->connPool[ctx->connToUse].conn, + if (edg_wll_plain_write_full(&ctx->connPlain->sock, buf, blen, &ctx->p_tmp_timeout) < 0) return edg_wll_SetError(ctx, errno, "edg_wll_http_send()"); } - if ( edg_wll_plain_write_full(&ctx->connPool[ctx->connToUse].conn, + if ( edg_wll_plain_write_full(&ctx->connPlain->sock, "\r\n", 2, &ctx->p_tmp_timeout) < 0) return edg_wll_SetError(ctx, errno, "edg_wll_http_send()"); - if ( body && edg_wll_plain_write_full(&ctx->connPool[ctx->connToUse].conn, + if ( body && edg_wll_plain_write_full(&ctx->connPlain->sock, body, len, &ctx->p_tmp_timeout) < 0) return edg_wll_SetError(ctx, errno, "edg_wll_http_send()"); diff --git a/org.glite.lb.common/src/xml_conversions.c b/org.glite.lb.common/src/xml_conversions.c index f14f0c8..515a6d6 100644 --- a/org.glite.lb.common/src/xml_conversions.c +++ b/org.glite.lb.common/src/xml_conversions.c @@ -58,6 +58,9 @@ void edg_wll_initXMLCtx(edg_wll_XML_ctx *c) { c->notifId = NULL; c->notifChangeOp = EDG_WLL_NOTIF_NOOP; c->notifValidity = -1; + c->jobId = NULL; + c->source = NULL; + c->seqCode = NULL; c->attrsGlobal = NULL; c->errCode = 0; c->bound = 0; @@ -834,3 +837,4 @@ char *edg_wll_NotifChangeOpToString(edg_wll_NotifChangeOp notifChangeOpConst) if (notifChangeOpConst < 0 || (notifChangeOpConst) > sizeof(notifChangeOpConsts)/sizeof(notifChangeOpConsts[0])) return (char *) NULL; return strdup(notifChangeOpConsts[(int) notifChangeOpConst]); } + diff --git a/org.glite.lb.common/src/xml_parse.c.T b/org.glite.lb.common/src/xml_parse.c.T index fec866d..0b7464f 100644 --- a/org.glite.lb.common/src/xml_parse.c.T +++ b/org.glite.lb.common/src/xml_parse.c.T @@ -25,26 +25,28 @@ #define UNUSED_VAR #endif -#define QUERY_EVENTS_REQUEST_BEGIN "\r\n" -#define QUERY_EVENTS_OREC_BEGIN "\t\t\r\n" -#define QUERY_EVENTS_OREC_END "\t\t\r\n" -#define QUERY_EVENTS_ORJC_BEGIN "\t\t\r\n" -#define QUERY_EVENTS_ORJC_END "\t\t\r\n" -#define QUERY_JOBS_REQUEST_BEGIN "\r\n" -#define QUERY_JOBS_OR_BEGIN "\t\t\r\n" -#define QUERY_JOBS_OR_END "\t\t\r\n" -#define PURGE_REQUEST_BEGIN "\r\n" -#define PURGE_REQUEST_END "\r\n" -#define DUMP_REQUEST_BEGIN "\r\n" -#define DUMP_REQUEST_END "\r\n" -#define LOAD_REQUEST_BEGIN "\r\n" -#define LOAD_REQUEST_END "\r\n" -#define INDEXED_ATTRS_REQUEST_BEGIN "\r\n" -#define INDEXED_ATTRS_REQUEST_END "\r\n" -#define NOTIF_REQUEST_BEGIN "\r\n" +#define QUERY_EVENTS_REQUEST_BEGIN "\r\n" +#define QUERY_EVENTS_OREC_BEGIN "\t\t\r\n" +#define QUERY_EVENTS_OREC_END "\t\t\r\n" +#define QUERY_EVENTS_ORJC_BEGIN "\t\t\r\n" +#define QUERY_EVENTS_ORJC_END "\t\t\r\n" +#define QUERY_JOBS_REQUEST_BEGIN "\r\n" +#define QUERY_JOBS_OR_BEGIN "\t\t\r\n" +#define QUERY_JOBS_OR_END "\t\t\r\n" +#define PURGE_REQUEST_BEGIN "\r\n" +#define PURGE_REQUEST_END "\r\n" +#define DUMP_REQUEST_BEGIN "\r\n" +#define DUMP_REQUEST_END "\r\n" +#define LOAD_REQUEST_BEGIN "\r\n" +#define LOAD_REQUEST_END "\r\n" +#define INDEXED_ATTRS_REQUEST_BEGIN "\r\n" +#define INDEXED_ATTRS_REQUEST_END "\r\n" +#define NOTIF_REQUEST_BEGIN "\r\n" +#define QUERY_SEQUENCE_CODE_REQUEST_BEGIN "\r\n" +#define QUERY_SEQUENCE_CODE_REQUEST_END "\r\n" /* lists of accepted tags */ @@ -660,6 +662,34 @@ static void startNotifResult(void *data, const char *el, const char **attr) +static void startQuerySequenceCodeResult(void *data, const char *el, const char **attr) +{ + edg_wll_XML_ctx *XMLCtx = data; + int i; + + + strcpy(XMLCtx->element, el); + + switch (XMLCtx->level) { + case 0: if (strcasecmp(el,"edg_wll_QuerySequenceCodeResult")) { unexpError() break;} + for ( i = 0; attr[i] && attr[i+1]; i += 2 ) { + if (!strcmp(attr[i],"code")) + XMLCtx->errCode = atoi(attr[i+1]); + else if (!strcmp(attr[i],"desc")) + XMLCtx->errDesc = strdup(attr[i+1]); + else { unexpError() } + } + break; + case 1: if (strcasecmp(el,"sequence_code")) unexpWarning() + break; + default: unexpWarning() + break; + } + XMLCtx->level++; +} + + + static void char_handler(void *data, const char *s, int len) { edg_wll_XML_ctx *XMLCtx = data; @@ -1199,6 +1229,25 @@ static void endNotifResult(void *data, const char *el UNUSED_VAR) } + +static void endQuerySequenceCodeResult(void *data, const char *el UNUSED_VAR) +{ + edg_wll_XML_ctx *XMLCtx = data; + + if (XMLCtx->level == 2) { + if (!strcmp(XMLCtx->element,"sequence_code")) + XMLCtx->seqCode = edg_wll_from_string_to_string(XMLCtx); + } + + XMLCtx->char_buf = NULL; + XMLCtx->char_buf_len = 0; + XMLCtx->level--; +} + + + + + #undef unexpError #undef unexpWarning @@ -2164,6 +2213,67 @@ edg_wll_ErrorCode edg_wll_ParseNotifResult(edg_wll_Context ctx, char *messageBod +/* parse sequence code result from server */ +edg_wll_ErrorCode edg_wll_ParseQuerySequenceCodeResult(edg_wll_Context ctx, char *messageBody, char **seqCode) +{ + edg_wll_XML_ctx XMLCtx; + edg_wll_ErrorCode errorCode; + XML_Char *encoding = "ISO-8859-1"; + + errno = 0; + edg_wll_ResetError(ctx); + edg_wll_initXMLCtx(&XMLCtx); + XMLCtx.ctx = ctx; + + + /* initialize parser */ + XMLCtx.p = XML_ParserCreate(encoding); + XML_SetElementHandler(XMLCtx.p, startQuerySequenceCodeResult, endQuerySequenceCodeResult); + XML_SetCharacterDataHandler(XMLCtx.p, char_handler); + XML_SetUserData(XMLCtx.p, (void *) &XMLCtx); + + + if (! XML_Parse(XMLCtx.p, messageBody, strlen(messageBody), 1)) { + char *errorMessage; + + asprintf(&errorMessage, "Parse error at line %d:\n%s\n", + XML_GetCurrentLineNumber(XMLCtx.p), + XML_ErrorString(XML_GetErrorCode(XMLCtx.p))); + + edg_wll_SetError(ctx, EDG_WLL_ERROR_XML_PARSE, errorMessage); + free(errorMessage); + } else if (XMLCtx.errtxt) edg_wll_SetError(ctx, EDG_WLL_ERROR_XML_PARSE, XMLCtx.errtxt); + + + if ((errorCode = edg_wll_Error(ctx,NULL,NULL))) { + free(XMLCtx.seqCode); + *seqCode = NULL; + } else { + *seqCode = XMLCtx.seqCode; + } + + if (XMLCtx.errDesc || XMLCtx.errCode) { + ctx->errDesc = XMLCtx.errDesc; + ctx->errCode = XMLCtx.errCode; + } + + /* print all warning if corresponding env variable is set */ + if (XMLCtx.warntxt && getenv("EDG_WLL_XML_WARNINGS")) { + fprintf(stderr,"----------------------------------------------------\n"); + fprintf(stderr,"%s\n\n",XMLCtx.warntxt); + fprintf(stderr,"%s\n",messageBody); + fprintf(stderr,"----------------------------------------------------\n"); + } + + /* free parser */ + XML_ParserFree(XMLCtx.p); + + edg_wll_freeXMLCtx(&XMLCtx); + return errorCode; +} + + + /* construct Message-Body of Request-Line for edg_wll_QueryJobs */ int edg_wll_JobQueryRecToXML( @@ -2610,3 +2720,27 @@ int edg_wll_NotifRequestToXML( return 0; } + + +/* construct Message-Body of Request-Line for edg_wll_QuerySequeceCode function */ +int edg_wll_QuerySequenceCodeToXML( + edg_wll_Context ctx, + edg_wlc_JobId jobId, + char **message) +{ + char *pomA=NULL, *pomC=NULL; + + + pomA = strdup(""); + edg_wll_add_jobid_to_XMLBody(&pomA, jobId, "jobId", NULL); + edg_wll_add_string_to_XMLBody(&pomA, edg_wll_SourceToString(ctx->p_source), "source", NULL); + + trio_asprintf(&pomC,"%s%s%s", + QUERY_SEQUENCE_CODE_REQUEST_BEGIN,pomA,QUERY_SEQUENCE_CODE_REQUEST_END); + + + free(pomA); + *message = pomC; + + return 0; +} diff --git a/org.glite.lb.server/src/lb_proto.c b/org.glite.lb.server/src/lb_proto.c index 582c41b..b1241d2 100644 --- a/org.glite.lb.server/src/lb_proto.c +++ b/org.glite.lb.server/src/lb_proto.c @@ -34,6 +34,7 @@ #define KEY_LOAD_REQUEST "/loadRequest " #define KEY_INDEXED_ATTRS "/indexedAttrs " #define KEY_NOTIF_REQUEST "/notifRequest " +#define KEY_QUERY_SEQUENCE_CODE "/querySequenceCode " #define KEY_HTTP "HTTP/1.1" @@ -786,6 +787,46 @@ edg_wll_ErrorCode edg_wll_Proto(edg_wll_Context ctx, free(conditions); } } + else if (!strncmp(requestPTR,KEY_QUERY_SEQUENCE_CODE,sizeof(KEY_QUERY_SEQUENCE_CODE)-1)) { + char *source; + char *seqCode; + edg_wlc_JobId jobId; + + + if (parseQuerySequenceCodeRequest(ctx, messageBody, &jobId, &source)) + ret = HTTP_BADREQ; + else { + int fatal = 0; + +/* XXX - needs server-side function + * consult error codes with nykolas + * + switch (edg_wll_QuerySequenceCode(ctx, jobId, source, &seqCode)) { + case 0: if (html) ret = HTTP_NOTIMPL; + else ret = HTTP_OK; + break; + case EEXIST: ret = HTTP_OK; break; + case EINVAL: ret = HTTP_INVALID; break; + case ENOENT: ret = HTTP_NOTFOUND; break; + case EPERM : ret = HTTP_UNAUTH; break; + case EDG_WLL_ERROR_NOINDEX: ret = HTTP_UNAUTH; break; + case ENOMEM: fatal = 1; ret = HTTP_INTERNAL; break; + default: ret = HTTP_INTERNAL; break; + } +*/ + + /* glue errors (if eny) to XML responce */ + if (!html && !fatal) + if (edg_wll_QuerySequenceCodeResultToXML(ctx, seqCode, &message)) + ret = HTTP_INTERNAL; + } + + free(source); + free(seqCode); + edg_wlc_JobIdFree(jobId); + + } + /* POST [something else]: not understood */ else ret = HTTP_BADREQ; diff --git a/org.glite.lb.server/src/lb_xml_parse.c.T b/org.glite.lb.server/src/lb_xml_parse.c.T index 13254e4..6732a04 100644 --- a/org.glite.lb.server/src/lb_xml_parse.c.T +++ b/org.glite.lb.server/src/lb_xml_parse.c.T @@ -26,21 +26,22 @@ #define UNUSED_VAR #endif -#define QUERY_EVENTS_BEGIN "\r\n" -#define QUERY_JOBS_BEGIN "\r\n" -#define PURGE_RESULT_BEGIN "\r\n" -#define DUMP_RESULT_BEGIN "\r\n" -#define LOAD_RESULT_BEGIN "\r\n" -#define INDEXED_ATTRS_BEGIN "\r\n" -#define NOTIF_RESULT_BEGIN "\r\n" - +#define QUERY_EVENTS_BEGIN "\r\n" +#define QUERY_JOBS_BEGIN "\r\n" +#define PURGE_RESULT_BEGIN "\r\n" +#define DUMP_RESULT_BEGIN "\r\n" +#define LOAD_RESULT_BEGIN "\r\n" +#define INDEXED_ATTRS_BEGIN "\r\n" +#define NOTIF_RESULT_BEGIN "\r\n" +#define QUERY_SEQUENCE_CODE_RESULT_BEGIN "\r\n" // XXX will be redundant soon @@ -378,6 +379,26 @@ static void startNotifRequest(void *data, const char *el, const char **attr) XMLCtx->level++; } + + +static void startQuerySequenceCodeRequest(void *data, const char *el, const char **attr) +{ + edg_wll_XML_ctx *XMLCtx = data; + + + strcpy(XMLCtx->element, el); + + switch (XMLCtx->level) { + case 0: if (strcasecmp(el,"edg_wll_QuerySequenceCodeRequest")) unexp() + break; + case 1: if ( (strcasecmp(el,"jobId")) && (strcasecmp(el,"source")) ) unexp() + break; + default: unexp() + break; + } + XMLCtx->level++; +} + #undef unexp @@ -788,6 +809,27 @@ static void endNotifRequest(void *data, const char *el UNUSED_VAR) +static void endQuerySequenceCodeRequest(void *data, const char *el UNUSED_VAR) +{ + edg_wll_XML_ctx *XMLCtx = data; + + if (XMLCtx->level == 2) { + if (!strcmp(XMLCtx->element,"jobId")) { + XMLCtx->jobId = edg_wll_from_string_to_jobid(XMLCtx); + } + else if (!strcmp(XMLCtx->element,"source")) { + XMLCtx->source = edg_wll_from_string_to_string(XMLCtx); + } + } + + XMLCtx->char_buf = NULL; + XMLCtx->char_buf_len = 0; + XMLCtx->level--; +} + + + + int parseJobQueryRec(edg_wll_Context ctx, const char *messageBody, long len, edg_wll_QueryRec ***conditions) { int ret; @@ -1200,6 +1242,60 @@ int parseNotifRequest(edg_wll_Context ctx, char *messageBody, char **function, e } + +/* parse Sequence Code request from client */ +int parseQuerySequenceCodeRequest(edg_wll_Context ctx, char *messageBody, edg_wlc_JobId *jobId, char **source) +{ + int ret; + edg_wll_XML_ctx XMLCtx; + XML_Char *encoding = "ISO-8859-1"; + + errno = 0; + edg_wll_initXMLCtx(&XMLCtx); + XMLCtx.ctx = ctx; + edg_wll_ResetError(ctx); + + + /* initialize parser */ + XMLCtx.p = XML_ParserCreate(encoding); + XML_SetElementHandler(XMLCtx.p, startQuerySequenceCodeRequest, endQuerySequenceCodeRequest); + XML_SetCharacterDataHandler(XMLCtx.p, char_handler); + XML_SetUserData(XMLCtx.p, (void *) &XMLCtx); + + + if (! XML_Parse(XMLCtx.p, messageBody, strlen(messageBody), 1)) { + char *errorMessage; + + asprintf(&errorMessage, "Parse error at line %d:\n%s\n", + XML_GetCurrentLineNumber(XMLCtx.p), + XML_ErrorString(XML_GetErrorCode(XMLCtx.p))); + + edg_wll_SetError(ctx, EDG_WLL_ERROR_XML_PARSE, errorMessage); + free(errorMessage); + } else if (XMLCtx.errtxt) edg_wll_SetError(ctx, EDG_WLL_ERROR_XML_PARSE, XMLCtx.errtxt); + + + if ((ret = edg_wll_Error(ctx,NULL,NULL))) { + if (XMLCtx.jobId) + edg_wlc_JobIdFree(XMLCtx.jobId); + + *jobId = NULL; + *source = NULL; + } + else { + *jobId = XMLCtx.jobId; + *source = XMLCtx.source; + } + + + XML_ParserFree(XMLCtx.p); + edg_wll_freeXMLCtx(&XMLCtx); + return ret; +} + + + + int edg_wll_QueryEventsToXML(edg_wll_Context ctx, edg_wll_Event *eventsOut, char **message) { char *pomA, *pomB; @@ -1729,3 +1825,30 @@ int edg_wll_NotifResultToXML( } + +/* construct Message-Body of Request-Line for edg_wll_Notif */ +int edg_wll_QuerySequenceCodeResultToXML( + edg_wll_Context ctx, + char *seqCode, + char **message) +{ + char *pomA, *pomB; + + + pomA = strdup(""); + edg_wll_add_string_to_XMLBody(&pomA, seqCode, "sequence_code", NULL); + + if (ctx->errDesc || ctx->errCode) + trio_asprintf(&pomB,"%s code=\"%d\" desc=\"%|Xs\">\r\n%s%s", + QUERY_SEQUENCE_CODE_RESULT_BEGIN, ctx->errCode, + ctx->errDesc, pomA, QUERY_SEQUENCE_CODE_RESULT_END); + else + trio_asprintf(&pomB,"%s>\r\n%s%s", QUERY_SEQUENCE_CODE_RESULT_BEGIN, + pomA, QUERY_SEQUENCE_CODE_RESULT_END); + free(pomA); + + *message = pomB; + return 0; +} + + diff --git a/org.glite.lb.server/src/lb_xml_parse.h b/org.glite.lb.server/src/lb_xml_parse.h index 587d150..539223e 100644 --- a/org.glite.lb.server/src/lb_xml_parse.h +++ b/org.glite.lb.server/src/lb_xml_parse.h @@ -22,6 +22,7 @@ int parsePurgeRequest(edg_wll_Context ctx, char *messageBody, int (*tagToIndex)( int parseDumpRequest(edg_wll_Context ctx, char *messageBody, edg_wll_DumpRequest *request); int parseLoadRequest(edg_wll_Context ctx, char *messageBody, edg_wll_LoadRequest *request); int parseNotifRequest(edg_wll_Context ctx, char *messageBody, char **function, edg_wll_NotifId *notifId, char **address, edg_wll_NotifChangeOp *op, edg_wll_QueryRec ***conditions); +int parseQuerySequenceCodeRequest(edg_wll_Context ctx, char *messageBody, edg_wlc_JobId *jobId, char **source); int edg_wll_QueryEventsToXML(edg_wll_Context, edg_wll_Event *, char **); int edg_wll_QueryJobsToXML(edg_wll_Context, edg_wlc_JobId *, edg_wll_JobStat *, char **); int edg_wll_JobStatusToXML(edg_wll_Context, edg_wll_JobStat, char **); @@ -31,6 +32,7 @@ int edg_wll_DumpResultToXML(edg_wll_Context ctx, edg_wll_DumpResult *result, cha int edg_wll_LoadResultToXML(edg_wll_Context ctx, edg_wll_LoadResult *result, char **message); int edg_wll_IndexedAttrsToXML(edg_wll_Context ctx, char **message); int edg_wll_NotifResultToXML(edg_wll_Context ctx, time_t validity, char **message); +int edg_wll_QuerySequenceCodeResultToXML(edg_wll_Context ctx, char *source, char **message); #ifdef __cplusplus }