<s> from nose . tools import assert_raises <EOL> import gevent <EOL> import sys <EOL> from zerorpc import zmq <EOL> import zerorpc <EOL> from testutils import teardown , random_ipc_endpoint , TIME_FACTOR <EOL> def test_close_server_hbchan ( ) : <EOL> endpoint = random_ipc_endpoint ( ) <EOL> server_events = zerorpc . Events ( zmq . ROUTER ) <EOL> server_events . bind ( endpoint ) <EOL> server = zerorpc . ChannelMultiplexer ( server_events ) <EOL> client_events = zerorpc . Events ( zmq . DEALER ) <EOL> client_events . connect ( endpoint ) <EOL> client = zerorpc . ChannelMultiplexer ( client_events , ignore_broadcast = True ) <EOL> client_channel = client . channel ( ) <EOL> client_hbchan = zerorpc . HeartBeatOnChannel ( client_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> client_hbchan . emit ( '<STR_LIT>' , None ) <EOL> event = server . recv ( ) <EOL> server_channel = server . channel ( event ) <EOL> server_hbchan = zerorpc . HeartBeatOnChannel ( server_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> server_hbchan . recv ( ) <EOL> gevent . sleep ( TIME_FACTOR * <NUM_LIT:3> ) <EOL> print '<STR_LIT>' <EOL> server_hbchan . close ( ) <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) : <EOL> assert_raises ( zerorpc . LostRemote , client_hbchan . recv ) <EOL> else : <EOL> with assert_raises ( zerorpc . LostRemote ) : <EOL> client_hbchan . recv ( ) <EOL> print '<STR_LIT>' <EOL> client_hbchan . close ( ) <EOL> server . close ( ) <EOL> client . close ( ) <EOL> def test_close_client_hbchan ( ) : <EOL> endpoint = random_ipc_endpoint ( ) <EOL> server_events = zerorpc . Events ( zmq . ROUTER ) <EOL> server_events . bind ( endpoint ) <EOL> server = zerorpc . ChannelMultiplexer ( server_events ) <EOL> client_events = zerorpc . Events ( zmq . DEALER ) <EOL> client_events . connect ( endpoint ) <EOL> client = zerorpc . ChannelMultiplexer ( client_events , ignore_broadcast = True ) <EOL> client_channel = client . channel ( ) <EOL> client_hbchan = zerorpc . HeartBeatOnChannel ( client_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> client_hbchan . emit ( '<STR_LIT>' , None ) <EOL> event = server . recv ( ) <EOL> server_channel = server . channel ( event ) <EOL> server_hbchan = zerorpc . HeartBeatOnChannel ( server_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> server_hbchan . recv ( ) <EOL> gevent . sleep ( TIME_FACTOR * <NUM_LIT:3> ) <EOL> print '<STR_LIT>' <EOL> client_hbchan . close ( ) <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) : <EOL> assert_raises ( zerorpc . LostRemote , server_hbchan . recv ) <EOL> else : <EOL> with assert_raises ( zerorpc . LostRemote ) : <EOL> server_hbchan . recv ( ) <EOL> print '<STR_LIT>' <EOL> server_hbchan . close ( ) <EOL> server . close ( ) <EOL> client . close ( ) <EOL> def test_heartbeat_can_open_channel_server_close ( ) : <EOL> endpoint = random_ipc_endpoint ( ) <EOL> server_events = zerorpc . Events ( zmq . ROUTER ) <EOL> server_events . bind ( endpoint ) <EOL> server = zerorpc . ChannelMultiplexer ( server_events ) <EOL> client_events = zerorpc . Events ( zmq . DEALER ) <EOL> client_events . connect ( endpoint ) <EOL> client = zerorpc . ChannelMultiplexer ( client_events , ignore_broadcast = True ) <EOL> client_channel = client . channel ( ) <EOL> client_hbchan = zerorpc . HeartBeatOnChannel ( client_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> event = server . recv ( ) <EOL> server_channel = server . channel ( event ) <EOL> server_hbchan = zerorpc . HeartBeatOnChannel ( server_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> gevent . sleep ( TIME_FACTOR * <NUM_LIT:3> ) <EOL> print '<STR_LIT>' <EOL> server_hbchan . close ( ) <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) : <EOL> assert_raises ( zerorpc . LostRemote , client_hbchan . recv ) <EOL> else : <EOL> with assert_raises ( zerorpc . LostRemote ) : <EOL> client_hbchan . recv ( ) <EOL> print '<STR_LIT>' <EOL> client_hbchan . close ( ) <EOL> server . close ( ) <EOL> client . close ( ) <EOL> def test_heartbeat_can_open_channel_client_close ( ) : <EOL> endpoint = random_ipc_endpoint ( ) <EOL> server_events = zerorpc . Events ( zmq . ROUTER ) <EOL> server_events . bind ( endpoint ) <EOL> server = zerorpc . ChannelMultiplexer ( server_events ) <EOL> client_events = zerorpc . Events ( zmq . DEALER ) <EOL> client_events . connect ( endpoint ) <EOL> client = zerorpc . ChannelMultiplexer ( client_events , ignore_broadcast = True ) <EOL> client_channel = client . channel ( ) <EOL> client_hbchan = zerorpc . HeartBeatOnChannel ( client_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> event = server . recv ( ) <EOL> server_channel = server . channel ( event ) <EOL> server_hbchan = zerorpc . HeartBeatOnChannel ( server_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> gevent . sleep ( TIME_FACTOR * <NUM_LIT:3> ) <EOL> print '<STR_LIT>' <EOL> client_hbchan . close ( ) <EOL> client . close ( ) <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) : <EOL> assert_raises ( zerorpc . LostRemote , server_hbchan . recv ) <EOL> else : <EOL> with assert_raises ( zerorpc . LostRemote ) : <EOL> server_hbchan . recv ( ) <EOL> print '<STR_LIT>' <EOL> server_hbchan . close ( ) <EOL> server . close ( ) <EOL> def test_do_some_req_rep ( ) : <EOL> endpoint = random_ipc_endpoint ( ) <EOL> server_events = zerorpc . Events ( zmq . ROUTER ) <EOL> server_events . bind ( endpoint ) <EOL> server = zerorpc . ChannelMultiplexer ( server_events ) <EOL> client_events = zerorpc . Events ( zmq . DEALER ) <EOL> client_events . connect ( endpoint ) <EOL> client = zerorpc . ChannelMultiplexer ( client_events , ignore_broadcast = True ) <EOL> client_channel = client . channel ( ) <EOL> client_hbchan = zerorpc . HeartBeatOnChannel ( client_channel , freq = TIME_FACTOR * <NUM_LIT:4> ) <EOL> event = server . recv ( ) <EOL> server_channel = server . channel ( event ) <EOL> server_hbchan = zerorpc . HeartBeatOnChannel ( server_channel , freq = TIME_FACTOR * <NUM_LIT:4> ) <EOL> def client_do ( ) : <EOL> for x in xrange ( <NUM_LIT:20> ) : <EOL> client_hbchan . emit ( '<STR_LIT>' , ( x , x * x ) ) <EOL> event = client_hbchan . recv ( ) <EOL> assert event . name == '<STR_LIT:OK>' <EOL> assert list ( event . args ) == [ x + x * x ] <EOL> client_hbchan . close ( ) <EOL> client_task = gevent . spawn ( client_do ) <EOL> def server_do ( ) : <EOL> for x in xrange ( <NUM_LIT:20> ) : <EOL> event = server_hbchan . recv ( ) <EOL> assert event . name == '<STR_LIT>' <EOL> server_hbchan . emit ( '<STR_LIT:OK>' , ( sum ( event . args ) , ) ) <EOL> server_hbchan . close ( ) <EOL> server_task = gevent . spawn ( server_do ) <EOL> server_task . get ( ) <EOL> client_task . get ( ) <EOL> client . close ( ) <EOL> server . close ( ) <EOL> def test_do_some_req_rep_lost_server ( ) : <EOL> endpoint = random_ipc_endpoint ( ) <EOL> server_events = zerorpc . Events ( zmq . ROUTER ) <EOL> server_events . bind ( endpoint ) <EOL> server = zerorpc . ChannelMultiplexer ( server_events ) <EOL> client_events = zerorpc . Events ( zmq . DEALER ) <EOL> client_events . connect ( endpoint ) <EOL> client = zerorpc . ChannelMultiplexer ( client_events , ignore_broadcast = True ) <EOL> def client_do ( ) : <EOL> print '<STR_LIT>' <EOL> client_channel = client . channel ( ) <EOL> client_hbchan = zerorpc . HeartBeatOnChannel ( client_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> for x in xrange ( <NUM_LIT:10> ) : <EOL> client_hbchan . emit ( '<STR_LIT>' , ( x , x * x ) ) <EOL> event = client_hbchan . recv ( ) <EOL> assert event . name == '<STR_LIT:OK>' <EOL> assert list ( event . args ) == [ x + x * x ] <EOL> client_hbchan . emit ( '<STR_LIT>' , ( x , x * x ) ) <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) : <EOL> assert_raises ( zerorpc . LostRemote , client_hbchan . recv ) <EOL> else : <EOL> with assert_raises ( zerorpc . LostRemote ) : <EOL> client_hbchan . recv ( ) <EOL> client_hbchan . close ( ) <EOL> client_task = gevent . spawn ( client_do ) <EOL> def server_do ( ) : <EOL> event = server . recv ( ) <EOL> server_channel = server . channel ( event ) <EOL> server_hbchan = zerorpc . HeartBeatOnChannel ( server_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> for x in xrange ( <NUM_LIT:10> ) : <EOL> event = server_hbchan . recv ( ) <EOL> assert event . name == '<STR_LIT>' <EOL> server_hbchan . emit ( '<STR_LIT:OK>' , ( sum ( event . args ) , ) ) <EOL> server_hbchan . close ( ) <EOL> server_task = gevent . spawn ( server_do ) <EOL> server_task . get ( ) <EOL> client_task . get ( ) <EOL> client . close ( ) <EOL> server . close ( ) <EOL> def test_do_some_req_rep_lost_client ( ) : <EOL> endpoint = random_ipc_endpoint ( ) <EOL> server_events = zerorpc . Events ( zmq . ROUTER ) <EOL> server_events . bind ( endpoint ) <EOL> server = zerorpc . ChannelMultiplexer ( server_events ) <EOL> client_events = zerorpc . Events ( zmq . DEALER ) <EOL> client_events . connect ( endpoint ) <EOL> client = zerorpc . ChannelMultiplexer ( client_events , ignore_broadcast = True ) <EOL> def client_do ( ) : <EOL> client_channel = client . channel ( ) <EOL> client_hbchan = zerorpc . HeartBeatOnChannel ( client_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> for x in xrange ( <NUM_LIT:10> ) : <EOL> client_hbchan . emit ( '<STR_LIT>' , ( x , x * x ) ) <EOL> event = client_hbchan . recv ( ) <EOL> assert event . name == '<STR_LIT:OK>' <EOL> assert list ( event . args ) == [ x + x * x ] <EOL> client_hbchan . close ( ) <EOL> client_task = gevent . spawn ( client_do ) <EOL> def server_do ( ) : <EOL> event = server . recv ( ) <EOL> server_channel = server . channel ( event ) <EOL> server_hbchan = zerorpc . HeartBeatOnChannel ( server_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> for x in xrange ( <NUM_LIT:10> ) : <EOL> event = server_hbchan . recv ( ) <EOL> assert event . name == '<STR_LIT>' <EOL> server_hbchan . emit ( '<STR_LIT:OK>' , ( sum ( event . args ) , ) ) <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) : <EOL> assert_raises ( zerorpc . LostRemote , server_hbchan . recv ) <EOL> else : <EOL> with assert_raises ( zerorpc . LostRemote ) : <EOL> server_hbchan . recv ( ) <EOL> server_hbchan . close ( ) <EOL> server_task = gevent . spawn ( server_do ) <EOL> server_task . get ( ) <EOL> client_task . get ( ) <EOL> client . close ( ) <EOL> server . close ( ) <EOL> def test_do_some_req_rep_client_timeout ( ) : <EOL> endpoint = random_ipc_endpoint ( ) <EOL> server_events = zerorpc . Events ( zmq . ROUTER ) <EOL> server_events . bind ( endpoint ) <EOL> server = zerorpc . ChannelMultiplexer ( server_events ) <EOL> client_events = zerorpc . Events ( zmq . DEALER ) <EOL> client_events . connect ( endpoint ) <EOL> client = zerorpc . ChannelMultiplexer ( client_events , ignore_broadcast = True ) <EOL> def client_do ( ) : <EOL> client_channel = client . channel ( ) <EOL> client_hbchan = zerorpc . HeartBeatOnChannel ( client_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) : <EOL> def _do_with_assert_raises ( ) : <EOL> for x in xrange ( <NUM_LIT:10> ) : <EOL> client_hbchan . emit ( '<STR_LIT>' , ( x , ) ) <EOL> event = client_hbchan . recv ( timeout = TIME_FACTOR * <NUM_LIT:3> ) <EOL> assert event . name == '<STR_LIT:OK>' <EOL> assert list ( event . args ) == [ x ] <EOL> assert_raises ( zerorpc . TimeoutExpired , _do_with_assert_raises ) <EOL> else : <EOL> with assert_raises ( zerorpc . TimeoutExpired ) : <EOL> for x in xrange ( <NUM_LIT:10> ) : <EOL> client_hbchan . emit ( '<STR_LIT>' , ( x , ) ) <EOL> event = client_hbchan . recv ( timeout = TIME_FACTOR * <NUM_LIT:3> ) <EOL> assert event . name == '<STR_LIT:OK>' <EOL> assert list ( event . args ) == [ x ] <EOL> client_hbchan . close ( ) <EOL> client_task = gevent . spawn ( client_do ) <EOL> def server_do ( ) : <EOL> event = server . recv ( ) <EOL> server_channel = server . channel ( event ) <EOL> server_hbchan = zerorpc . HeartBeatOnChannel ( server_channel , freq = TIME_FACTOR * <NUM_LIT:2> ) <EOL> if sys . version_info < ( <NUM_LIT:2> , <NUM_LIT:7> ) : <EOL> def _do_with_assert_raises ( ) : <EOL> for x in xrange ( <NUM_LIT:20> ) : <EOL> event = server_hbchan . recv ( ) <EOL> assert event . name == '<STR_LIT>' <EOL> gevent . sleep ( TIME_FACTOR * event . args [ <NUM_LIT:0> ] ) <EOL> server_hbchan . emit ( '<STR_LIT:OK>' , event . args ) <EOL> assert_raises ( zerorpc . LostRemote , _do_with_assert_raises ) <EOL> else : <EOL> with assert_raises ( zerorpc . LostRemote ) : <EOL> for x in xrange ( <NUM_LIT:20> ) : <EOL> event = server_hbchan . recv ( ) <EOL> assert event . name == '<STR_LIT>' <EOL> gevent . sleep ( TIME_FACTOR * event . args [ <NUM_LIT:0> ] ) <EOL> server_hbchan . emit ( '<STR_LIT:OK>' , event . args ) <EOL> server_hbchan . close ( ) <EOL> server_task = gevent . spawn ( server_do ) <EOL> server_task . get ( ) <EOL> client_task . get ( ) <EOL> client . close ( ) <EOL> server . close ( ) </s>
