@ -30,6 +30,7 @@
# include "JsonRPC.h"
# include "JsonRPCEvents.h"
# include "AmSession.h"
# include "AmEventDispatcher.h"
# include "log.h"
@ -55,6 +56,13 @@ ev_io ev_accept;
ev_async JsonRPCServerLoop : : async_w ;
struct ev_loop * JsonRPCServerLoop : : loop = 0 ;
JsonRPCServerLoop * JsonRPCServerLoop : : _instance = NULL ;
RpcServerThreadpool JsonRPCServerLoop : : threadpool ;
std : : map < string , JsonrpcPeerConnection * > JsonRPCServerLoop : : connections ;
AmMutex JsonRPCServerLoop : : connections_mut ;
vector < JsonServerEvent * > JsonRPCServerLoop : : pending_events ;
AmMutex JsonRPCServerLoop : : pending_events_mut ;
JsonRPCServerLoop * JsonRPCServerLoop : : instance ( ) {
if ( _instance = = NULL ) {
@ -63,8 +71,7 @@ JsonRPCServerLoop* JsonRPCServerLoop::instance() {
return _instance ;
}
int
setnonblock ( int fd )
int setnonblock ( int fd )
{
int flags ;
@ -95,40 +102,37 @@ setnonblock(int fd)
// }
static void read_cb ( struct ev_loop * loop , struct ev_io * w , int revents ) {
struct JsonrpcNetstringsConnection * cli = ( ( struct JsonrpcNetstringsConnection * ) ( ( ( char * ) w ) -
offsetof ( JsonrpcNetstringsConnection , ev_read ) ) ) ;
struct JsonrpcNetstringsConnection * peer =
( ( struct JsonrpcNetstringsConnection * )
( ( ( char * ) w ) - offsetof ( JsonrpcNetstringsConnection , ev_read ) ) ) ;
DBG ( " read_cb in connection %p \n " , peer ) ;
// int r=0;
// char rbuff[1024];
if ( revents & EV_READ ) {
int res = cli - > netstringsRead ( ) ;
// read message - here in main server thread (more efficient for small messages)
int res = peer - > netstringsRead ( ) ;
switch ( res ) {
case JsonrpcNetstringsConnection : : CONTINUE :
ev_io_start ( loop , & cli - > ev_read ) ; return ;
ev_io_start ( loop , & peer - > ev_read ) ; return ;
case JsonrpcNetstringsConnection : : REMOVE : {
ev_io_stop ( EV_A_ w ) ;
// let event receivers know about broken connection
// todo: add connection id
if ( ! cli - > notificationReceiver . empty ( ) )
AmEventDispatcher : : instance ( ) - > post ( cli - > notificationReceiver ,
new JsonRpcConnectionEvent ( JsonRpcConnectionEvent : : DISCONNECT ) ) ;
if ( ! cli - > requestReceiver . empty ( ) )
AmEventDispatcher : : instance ( ) - > post ( cli - > requestReceiver ,
new JsonRpcConnectionEvent ( JsonRpcConnectionEvent : : DISCONNECT ) ) ;
for ( std : : map < std : : string , std : : string > : : iterator it =
cli - > replyReceivers . begin ( ) ; it ! = cli - > replyReceivers . end ( ) ; it + + ) {
AmEventDispatcher : : instance ( ) - > post ( it - > second ,
new JsonRpcConnectionEvent ( JsonRpcConnectionEvent : : DISCONNECT ) ) ;
}
delete cli ;
peer - > notifyDisconnect ( ) ;
JsonRPCServerLoop : : instance ( ) - > removeConnection ( peer - > id ) ;
delete peer ;
} return ;
case JsonrpcNetstringsConnection : : DISPATCH : {
ev_io_stop ( EV_A_ w ) ;
JsonRPCServerLoop : : dispatchServerEvent ( new JsonServerEvent ( cli ) ) ;
JsonRPCServerLoop : :
dispatchServerEvent ( new JsonServerEvent ( peer , JsonServerEvent : : StartReadLoop ) ) ;
} return ;
}
// todo: put into reader thread
//r=read(cli->fd,&rbuff,1024);
return ;
}
// put back to read loop
@ -140,13 +144,12 @@ static void read_cb(struct ev_loop *loop, struct ev_io *w, int revents) {
}
void JsonRPCServerLoop : : dispatchServerEvent ( AmEvent * ev ) {
instance( ) - > threadpool. dispatch ( ev ) ;
. dispatch ( ev ) ;
}
static void accept_cb ( struct ev_loop * loop , struct ev_io * w , int revents )
{
int client_fd ;
struct JsonrpcNetstringsConnection * a_client ;
struct sockaddr_in client_addr ;
socklen_t client_len = sizeof ( client_addr ) ;
client_fd = accept ( w - > fd , ( struct sockaddr * ) & client_addr , & client_len ) ;
@ -154,14 +157,19 @@ static void accept_cb(struct ev_loop *loop, struct ev_io *w, int revents)
return ;
}
a_client = new JsonrpcNetstringsConnection ( ) ;
a_client - > fd = client_fd ;
if ( setnonblock ( a_client - > fd ) < 0 ) {
string connection_id = JsonRPCServerLoop : : newConnectionId ( ) ;
JsonrpcNetstringsConnection * peer = new JsonrpcNetstringsConnection ( connection_id ) ;
peer - > fd = client_fd ;
if ( setnonblock ( peer - > fd ) < 0 ) {
delete peer ;
ERROR ( " failed to set client socket to non-blocking " ) ;
return ;
}
ev_io_init ( & a_client - > ev_read , read_cb , a_client - > fd , EV_READ ) ;
ev_io_start ( loop , & a_client - > ev_read ) ;
JsonRPCServerLoop : : registerConnection ( peer , connection_id ) ;
ev_io_init ( & peer - > ev_read , read_cb , peer - > fd , EV_READ ) ;
ev_io_start ( loop , & peer - > ev_read ) ;
}
static void async_cb ( EV_P_ ev_async * w , int revents )
@ -174,28 +182,102 @@ void JsonRPCServerLoop::_processEvents() {
}
void JsonRPCServerLoop : : process ( AmEvent * ev ) {
DBG ( " in server loop \n " ) ;
DBG ( " server loop - processing event\n " ) ;
JsonServerEvent * server_event = dynamic_cast < JsonServerEvent * > ( ev ) ;
if ( server_event = = NULL ) {
ERROR ( " wrong event type received \n " ) ;
return ;
}
JsonrpcNetstringsConnection * a_client = server_event - > conn ;
a_client - > resetRead ( ) ;
ev_io_init ( & a_client - > ev_read , read_cb , a_client - > fd , EV_READ ) ;
ev_io_start ( loop , & a_client - > ev_read ) ;
switch ( server_event - > event_id ) {
case JsonServerEvent : : StartReadLoop : {
JsonrpcNetstringsConnection * a_client = server_event - > conn ;
DBG ( " checking for pending events to connection %p/%s \n " ,
a_client , a_client - > id . c_str ( ) ) ;
pending_events_mut . lock ( ) ;
// (check whether event for that connection pending)
for ( vector < JsonServerEvent * > : : iterator it =
pending_events . begin ( ) ; it ! = pending_events . end ( ) ; it + + ) {
if ( ( * it ) - > connection_id = = a_client - > id ) {
// stop read loop
ev_io_stop ( loop , & a_client - > ev_read ) ;
JsonServerEvent * server_event = * it ;
pending_events . erase ( it ) ;
pending_events_mut . unlock ( ) ;
DBG ( " got pending event for connection '%s' \n " , a_client - > id . c_str ( ) ) ;
server_event - > conn = a_client ;
dispatchServerEvent ( server_event ) ;
return ;
}
}
pending_events_mut . unlock ( ) ;
DBG ( " no pending events for connection %p/%s, starting read loop \n " ,
a_client , a_client - > id . c_str ( ) ) ;
a_client - > resetRead ( ) ;
ev_io_init ( & a_client - > ev_read , read_cb , a_client - > fd , EV_READ ) ;
ev_io_start ( loop , & a_client - > ev_read ) ;
} ; break ;
case JsonServerEvent : : SendMessage : {
JsonServerSendMessageEvent * snd_msg_ev =
dynamic_cast < JsonServerSendMessageEvent * > ( server_event ) ;
if ( snd_msg_ev = = NULL ) {
ERROR ( " invalid SendMessage type event received \n " ) ;
return ;
}
JsonrpcPeerConnection * p_peer = getConnection ( snd_msg_ev - > connection_id ) ;
if ( p_peer = = NULL ) {
WARN ( " dropping message to inexistent/broken connection '%s' "
" (is_reply=%s, method=%s, id=%s, params='%s') " ,
snd_msg_ev - > connection_id . c_str ( ) , snd_msg_ev - > is_reply ? " true " : " false " ,
snd_msg_ev - > method . c_str ( ) , snd_msg_ev - > id . c_str ( ) ,
AmArg : : print ( snd_msg_ev - > params ) . c_str ( ) ) ;
return ;
}
JsonrpcNetstringsConnection * peer = dynamic_cast < JsonrpcNetstringsConnection * > ( p_peer ) ;
if ( NULL = = peer ) {
ERROR ( " invalid connection type \n " ) ; // todo: other transports
return ;
}
if ( ev_is_active ( & peer - > ev_read ) ) {
// ok, peer is in read loop, we can dispatch it to thread for sending
ev_io_stop ( EV_A_ & peer - > ev_read ) ;
JsonRPCServerLoop : :
dispatchServerEvent ( new JsonServerSendMessageEvent ( * snd_msg_ev , peer ) ) ;
} else {
// peer is being processed by thread, save event for later sending
pending_events_mut . lock ( ) ;
// (need to copy event here: original event is deleted when event processed)
pending_events . push_back ( new JsonServerSendMessageEvent ( * snd_msg_ev ) ) ;
size_t q_size = pending_events . size ( ) ;
pending_events_mut . unlock ( ) ;
DBG ( " queued event for connection %s (total %zd events pending) \n " ,
snd_msg_ev - > connection_id . c_str ( ) , q_size ) ;
}
} ; break ;
// todo: process remove connection event
default :
ERROR ( " unknown server event type received \n " ) ;
return ;
}
}
JsonRPCServerLoop : : JsonRPCServerLoop ( )
: AmEventQueue ( this )
{
loop = ev_default_loop ( 0 ) ;
// one thread is started here so that
// in app initialization code, there is already
// a server thread available to receive events
DBG ( " starting one server thread for startup requests... \n " ) ;
threadpool . addThreads ( 1 ) ;
}
@ -239,6 +321,7 @@ void JsonRPCServerLoop::run() {
ev_io_init ( & ev_accept , accept_cb , listen_fd , EV_READ ) ;
ev_io_start ( loop , & ev_accept ) ;
// async watcher to process our events in event loop
ev_async_init ( & async_w , async_cb ) ;
ev_async_start ( EV_A_ & async_w ) ;
@ -252,8 +335,29 @@ void JsonRPCServerLoop::on_stop() {
}
void JsonRPCServerLoop : : returnConnection ( JsonrpcNetstringsConnection * conn ) {
pending_events_mut . lock ( ) ;
// (check whether event for that connection pending)
DBG ( " checking %u pending events \n " , pending_events . size ( ) ) ;
for ( vector < JsonServerEvent * > : : iterator it =
pending_events . begin ( ) ; it ! = pending_events . end ( ) ; it + + ) {
DBG ( " %s vs %s \n " , ( * it ) - > connection_id . c_str ( ) , conn - > id . c_str ( ) ) ;
if ( ( * it ) - > connection_id = = conn - > id ) {
JsonServerEvent * server_event = * it ;
pending_events . erase ( it ) ;
pending_events_mut . unlock ( ) ;
DBG ( " got pending event for connection '%s' \n " , conn - > id . c_str ( ) ) ;
server_event - > conn = conn ;
dispatchServerEvent ( server_event ) ;
return ;
}
}
pending_events_mut . unlock ( ) ;
DBG ( " returning connection %p \n " , conn ) ;
instance ( ) - > postEvent ( new JsonServerEvent ( conn ) ) ;
instance ( ) - > postEvent ( new JsonServerEvent ( conn , JsonServerEvent : : StartReadLoop ) ) ;
ev_async_send ( loop , & async_w ) ;
}
@ -265,7 +369,8 @@ void JsonRPCServerLoop::execRpc(const string& evq_link,
int port , const string & method ,
AmArg & params ,
AmArg & ret ) {
JsonrpcNetstringsConnection * peer = new JsonrpcNetstringsConnection ( ) ;
string connection_id = newConnectionId ( ) ;
JsonrpcNetstringsConnection * peer = new JsonrpcNetstringsConnection ( connection_id ) ;
peer - > flags = flags ;
peer - > notificationReceiver = notificationReceiver ;
peer - > requestReceiver = requestReceiver ;
@ -279,13 +384,101 @@ void JsonRPCServerLoop::execRpc(const string& evq_link,
return ;
}
if ( JsonRpcServer : : createRequest ( evq_link , method , params , peer ) ) {
// DBG("evq_link = '%s'\n", evq_link.c_str());
// if (JsonRpcServer::createRequest(evq_link, method, params, peer)) {
// ret.push(400);
// ret.push("Error creating request message");
// delete peer;
// return;
// }
registerConnection ( peer , connection_id ) ;
DBG ( " dispatching JsonServerSendMessageEvent \n " ) ;
JsonServerSendMessageEvent * send_message_event =
new JsonServerSendMessageEvent ( connection_id , false , method , " 1 " /* id - not empty */ ,
params ) ;
send_message_event - > reply_link = evq_link ;
JsonRPCServerLoop : : dispatchServerEvent ( send_message_event ) ;
// JsonRPCServerLoop::
// dispatchServerEvent(new JsonServerEvent(peer, JsonServerEvent::SendMessage));
ret . push ( 200 ) ;
ret . push ( " OK " ) ;
ret . push ( connection_id ) ;
}
void JsonRPCServerLoop : : sendMessage ( const string & connection_id ,
int msg_type ,
const string & method ,
const string & id ,
const string & reply_sink ,
AmArg & params ,
AmArg & ret ) {
// check for presence of connection
// (connection might still be removed until we really
// process the request to send message, but here we already
// catch most failures)
if ( getConnection ( connection_id ) = = NULL ) {
ret . push ( 400 ) ;
ret . push ( " Error creating request message " ) ;
ret . push ( " unknown connection " ) ;
return ;
}
JsonRPCServerLoop : : dispatchServerEvent ( new JsonServerEvent ( peer ) ) ;
JsonServerSendMessageEvent * ev =
new JsonServerSendMessageEvent ( connection_id , msg_type ! = JSONRPC_MSG_REQUEST ,
method , id , params , reply_sink ) ;
ev - > is_error = msg_type = = JSONRPC_MSG_ERROR ;
instance ( ) - > postEvent ( ev ) ;
// wake up event loop to process message
ev_async_send ( loop , & async_w ) ;
ret . push ( 200 ) ;
ret . push ( " OK " ) ;
ret . push ( " posted " ) ;
}
string JsonRPCServerLoop : : newConnectionId ( ) {
return AmSession : : getNewId ( ) ;
}
bool JsonRPCServerLoop : : registerConnection ( JsonrpcPeerConnection * peer , const string & id ) {
bool res = false ;
connections_mut . lock ( ) ;
if ( connections . find ( id ) ! = connections . end ( ) )
res = true ;
connections [ id ] = peer ;
connections_mut . unlock ( ) ;
DBG ( " registered connection '%s' \n " , id . c_str ( ) ) ;
return res ;
}
bool JsonRPCServerLoop : : removeConnection ( const string & id ) {
bool res = false ;
connections_mut . lock ( ) ;
std : : map < string , JsonrpcPeerConnection * > : : iterator it =
connections . find ( id ) ;
if ( it ! = connections . end ( ) ) {
res = true ;
connections . erase ( it ) ;
}
connections_mut . unlock ( ) ;
DBG ( " deregistered connection '%s' \n " , id . c_str ( ) ) ;
return res ;
}
JsonrpcPeerConnection * JsonRPCServerLoop : : getConnection ( const string & id ) {
JsonrpcPeerConnection * res = NULL ;
connections_mut . lock ( ) ;
std : : map < string , JsonrpcPeerConnection * > : : iterator it =
connections . find ( id ) ;
if ( it ! = connections . end ( ) )
res = it - > second ;
connections_mut . unlock ( ) ;
return res ;
}
Stefan Sayer
16 years ago