}
if( (whm = malloc( sizeof( *whm ) )) == NULL ) {
- fprintf( stderr, "mem alloc failed for whm: alloc %d bytes\n", (int) sizeof( *whm ) );
+ rmr_vlog( RMR_VL_ERR, "mem alloc failed for whm: alloc %d bytes\n", (int) sizeof( *whm ) );
errno = ENOMEM;
return 0;
}
whm->nalloc = 16;
alloc_sz = whm->nalloc * sizeof( endpoint_t );
if( (whm->eps = (endpoint_t **) malloc( alloc_sz )) == NULL ) {
- fprintf( stderr, "mem alloc failed: alloc %d bytes\n", (int) alloc_sz );
+ rmr_vlog( RMR_VL_ERR, "mem alloc failed: alloc %d bytes\n", (int) alloc_sz );
free( whm );
errno = ENOMEM;
return 0;
if( (ep = rt_ensure_ep( ctx->rtable, target )) == NULL ) { // get pointer to ep if there, create new if not
- fprintf( stderr, "ensure ep returned bad; setting no memory error\n" );
+ rmr_vlog( RMR_VL_ERR, "wormhole_open: ensure ep returned bad: target=(%s)\n", target );
return -1; // ensure sets errno
}
}
if( whm->eps[i] == ep ) {
- return i; // we're already pointing to it, just send it back again
+ if( whm->eps[i]->open ) { // we know about it and it's open
+ return i; // just send back the reference
+ }
+
+ whid = i; // have it, but not open, reopen
+ break;
}
}
}
}
- if( !rt_link2_ep( ep ) ) { // start a connection if open already
+ if( !rt_link2_ep( ctx, ep ) ) { // start a connection if not already open
errno = ECONNREFUSED;
return -1;
}
uta_ctx_t* ctx;
endpoint_t* ep; // enpoint that wormhole ID references
wh_mgt_t *whm;
+ char* d1; // point at the call-id in the header
if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) { // bad stuff, bail fast
errno = EINVAL; // if msg is null, this is their clue
return msg;
}
- errno = 0; // nng seems not to set errno any longer, so ensure it's clear
+ errno = 0;
if( msg->header == NULL ) {
- fprintf( stderr, "[ERR] rmr_wh_send_msg: message had no header\n" );
+ rmr_vlog( RMR_VL_ERR, "rmr_wh_send_msg: message had no header\n" );
msg->state = RMR_ERR_NOHDR;
errno = EBADMSG; // must ensure it's not eagain
return msg;
}
+ d1 = DATA1_ADDR( msg->header );
+ d1[D1_CALLID_IDX] = NO_CALL_ID; // must blot out so it doesn't queue on a chute at the other end
+
ep = whm->eps[whid];
+ if( ! ep->open ) {
+ rmr_wh_open( ctx, ep->name );
+ }
return send2ep( ctx, ep, msg ); // send directly to the endpoint
}
+/*
+ Send a message directly to an open wormhole and then block until a response has
+ been received. The return is the same as for rmr_call(); the received buffer
+ or nil if no response was received.
+*/
+extern rmr_mbuf_t* rmr_wh_call( void* vctx, rmr_whid_t whid, rmr_mbuf_t* msg, int call_id, int max_wait ) {
+ uta_ctx_t* ctx;
+ endpoint_t* ep; // enpoint that wormhole ID references
+ wh_mgt_t *whm;
+ char* d1; // point at the call-id in the header
+
+ if( (ctx = (uta_ctx_t *) vctx) == NULL || msg == NULL ) { // bad stuff, bail fast
+ errno = EINVAL; // if msg is null, this is their clue
+ if( msg != NULL ) {
+ msg->state = RMR_ERR_BADARG;
+ errno = EINVAL; // must ensure it's not eagain
+ }
+ return msg;
+ }
+
+ msg->state = RMR_OK;
+
+ if( (whm = ctx->wormholes) == NULL ) {
+ errno = EINVAL; // no wormholes open
+ msg->state = RMR_ERR_NOWHOPEN;
+ return msg;
+ }
+
+ if( whid < 0 || whid >= whm->nalloc || whm->eps[whid] == NULL ) {
+ errno = EINVAL;
+ msg->state = RMR_ERR_WHID;
+ return msg;
+ }
+
+ errno = 0;
+ if( msg->header == NULL ) {
+ rmr_vlog( RMR_VL_ERR, "rmr_wh_call: message had no header\n" );
+ msg->state = RMR_ERR_NOHDR;
+ errno = EBADMSG; // must ensure it's not eagain
+ return msg;
+ }
+
+ ep = whm->eps[whid];
+ if( ep != NULL ) {
+ if( ! ep->open ) {
+ rmr_wh_open( ctx, ep->name );
+ }
+ return mt_call( vctx, msg, call_id, max_wait, ep ); // use main (internal) call to setup and block
+ }
+
+ msg->state = RMR_ERR_NOENDPT;
+ return msg;
+}
+
/*
This will "close" a wormhole. We don't actually drop the session as that might be needed
by others, but we do pull the ep reference from the list.
whm->eps[whid] = NULL;
}
+
+/*
+ Check the state of an endpoint that is associated with the wormhold ID
+ passed in. If the state is "open" then we return RMR_OK. Other possible
+ return codes:
+
+ RMR_ERR_WHID // wormhole id was invalid
+ RMR_ERR_NOENDPT // the endpoint connection is not open
+ RMR_ERR_BADARG // context or other arg was invalid
+ RMR_ERR_NOWHOPEN // wormhole(s) have not been initalised
+
+*/
+extern int rmr_wh_state( void* vctx, rmr_whid_t whid ) {
+ uta_ctx_t* ctx;
+ wh_mgt_t* whm; // easy reference to wh mgt stuff
+ endpoint_t* ep; // enpoint that wormhole ID references
+
+ if( (ctx = (uta_ctx_t *) vctx) == NULL ) { // bad stuff, bail fast
+ errno = EINVAL;
+ return RMR_ERR_BADARG;
+ }
+
+ if( (whm = ctx->wormholes) == NULL ) {
+ errno = EINVAL; // no wormholes open
+ return RMR_ERR_NOWHOPEN;
+ }
+
+ if( whid < 0 || whid >= whm->nalloc || whm->eps[whid] == NULL ) {
+ errno = EINVAL;
+ return RMR_ERR_WHID;
+ }
+
+ errno = 0;
+
+ if( (ep = whm->eps[whid]) != NULL ) {
+ return ep->open ? RMR_OK : RMR_ERR_NOENDPT;
+ }
+
+ return RMR_ERR_NOENDPT;
+}