1818import os
1919import subprocess
2020import sys
21+ import time
2122
2223class ParseOptions ():
2324 """Parse arguments."""
@@ -93,9 +94,106 @@ def __parse_args(self):
9394 type = str ,
9495 help = 'Path to the xml configuration file containing discovery server.'
9596 )
97+ parser .add_argument (
98+ '-nc' ,
99+ '--n-clients' ,
100+ type = int ,
101+ help = 'Number of pubsub clients to launch (1 of each).'
102+ )
103+ parser .add_argument (
104+ '-rc' ,
105+ '--relaunch-clients' ,
106+ action = 'store_true' ,
107+ help = 'Whether to kill clients and relaunch them.'
108+ )
109+ parser .add_argument (
110+ '-vm' ,
111+ '--validation-method' ,
112+ type = str ,
113+ help = 'Validation method to use [server, subscriber].'
114+ )
115+ parser .add_argument (
116+ '-edr' ,
117+ '--exit-on-disposal-received' ,
118+ action = 'store_true' ,
119+ help = 'Let the publisher finish the process if receives a disposal.'
120+ )
96121
97122 return parser .parse_args ()
98123
124+ def launch_discovery_server_processes (servers , xml_servers ):
125+
126+ ds_procs = []
127+
128+ for i in range (0 , len (servers )):
129+ server_cmd = []
130+
131+ if not os .path .isfile (servers [i ]):
132+ print (f'Discovery server executable file does not exists: { servers [i ]} )
133+ sys .exit (1 )
134+
135+ if not os .access (servers [i ], os .X_OK ):
136+ print (
137+ 'Discovery server executable does not have execution permissions:'
138+ f'{ servers [i ]} )
139+ sys .exit (1 )
140+
141+ server_cmd .append (servers [i ])
142+ server_cmd .extend (['--xml-file' , xml_servers [i ]])
143+ server_cmd .extend (['--server-id' , str (i )])
144+
145+ ds_proc = subprocess .Popen (server_cmd )
146+ print (
147+ 'Running Discovery Server - commmand: ' ,
148+ ' ' .join (map (str , server_cmd )))
149+
150+ ds_procs .append (ds_proc )
151+
152+ return ds_procs
153+
154+ def launch_client_processes (n_clients , pub_command , sub_command ):
155+
156+ pub_procs = []
157+ sub_procs = []
158+
159+ for i in range (0 , n_clients ):
160+ sub_proc = subprocess .Popen (sub_command )
161+ print (
162+ f'Running Subscriber - commmand: ' ,
163+ ' ' .join (map (str , sub_command )))
164+
165+ pub_proc = subprocess .Popen (pub_command )
166+ print (
167+ 'Running Publisher - commmand: ' ,
168+ ' ' .join (map (str , pub_command )))
169+
170+ sub_procs .append (sub_proc )
171+ pub_procs .append (pub_proc )
172+
173+ return sub_procs , pub_procs
174+
175+ def cleanup (pub_procs , sub_procs , ds_procs ):
176+
177+ [sub_proc .kill () for sub_proc in sub_procs ]
178+ [pub_proc .kill () for pub_proc in pub_procs ]
179+ [ds_proc .kill () for ds_proc in ds_procs ]
180+
181+ def cleanup_clients (pub_procs , sub_procs ):
182+
183+ [sub_proc .kill () for sub_proc in sub_procs ]
184+ [pub_proc .kill () for pub_proc in pub_procs ]
185+
186+ def terminate (ok = True ):
187+ if ok :
188+ try :
189+ sys .exit (os .EX_OK )
190+ except AttributeError :
191+ sys .exit (0 )
192+ else :
193+ try :
194+ sys .exit (os .EX_SOFTWARE )
195+ except AttributeError :
196+ sys .exit (1 )
99197
100198def run (args ):
101199 """
@@ -108,6 +206,8 @@ def run(args):
108206 """
109207 pub_command = []
110208 sub_command = []
209+ n_clients = 1
210+ relaunch_clients = False
111211
112212 script_dir = os .path .dirname (os .path .realpath (__file__ ))
113213
@@ -152,71 +252,60 @@ def run(args):
152252 if args .wait :
153253 pub_command .extend (['--wait' , str (args .wait )])
154254
255+ if args .exit_on_disposal_received :
256+ pub_command .extend (['--exit_on_disposal_received' ])
257+
155258 if args .samples :
156259 pub_command .extend (['--samples' , str (args .samples )])
157260 sub_command .extend (['--samples' , str (args .samples )])
158261
262+ if args .n_clients :
263+ n_clients = int (args .n_clients )
264+
265+ if args .relaunch_clients :
266+ relaunch_clients = True
267+
159268 if len (args .servers ) != len (args .xml_servers ):
160269 print (
161270 'Number of servers arguments should be equal to the number of xmls provided.' )
162271 sys .exit (1 )
163272
164- ds_procs = []
165- for i in range (0 , len (args .servers )):
166- server_cmd = []
273+ ds_procs = launch_discovery_server_processes (args .servers , args .xml_servers )
167274
168- if not os .path .isfile (args .servers [i ]):
169- print (f'Discovery server executable file does not exists: { args .servers [i ]} )
170- sys .exit (1 )
275+ sub_procs , pub_procs = launch_client_processes (n_clients , pub_command , sub_command )
171276
172- if not os .access (args .servers [i ], os .X_OK ):
173- print (
174- 'Discovery server executable does not have execution permissions:'
175- f'{ args .servers [i ]} )
176- sys .exit (1 )
277+ terminate_ok = True
177278
178- server_cmd .append (args .servers [i ])
179- server_cmd .extend (['--xml-file' , args .xml_servers [i ]])
180- server_cmd .extend (['--server-id' , str (i )])
279+ if relaunch_clients :
280+ time .sleep (3 )
181281
182- ds_proc = subprocess .Popen (server_cmd )
183- print (
184- 'Running Discovery Server - commmand: ' ,
185- ' ' .join (map (str , server_cmd )))
282+ cleanup_clients (pub_procs , sub_procs )
283+ sub_procs , pub_procs = launch_client_processes (n_clients , pub_command , sub_command )
186284
187- ds_procs . append ( ds_proc )
285+ time . sleep ( 3 )
188286
189- sub_proc = subprocess .Popen (sub_command )
190- print (
191- f'Running Subscriber - commmand: ' ,
192- ' ' .join (map (str , sub_command )))
193-
194- pub_proc = subprocess .Popen (pub_command )
195- print (
196- 'Running Publisher - commmand: ' ,
197- ' ' .join (map (str , pub_command )))
198-
199- try :
200- outs , errs = sub_proc .communicate (timeout = 15 )
201- except subprocess .TimeoutExpired :
202- print ('Subscriber process timed out, terminating...' )
203- sub_proc .kill ()
204- pub_proc .kill ()
205- [ds_proc .kill () for ds_proc in ds_procs ]
287+ if args .validation_method == 'server' :
288+ # Check If discovery servers are still running
289+ for ds_proc in ds_procs :
290+ retcode = ds_proc .poll ()
291+ if retcode is not None and retcode is not 0 :
292+ print ('Discovery Server process dead, terminating...' )
293+ terminate_ok = False
294+ else :
206295 try :
207- sys .exit (os .EX_SOFTWARE )
208- except AttributeError :
209- sys .exit (1 )
296+ for sub_proc in sub_procs :
297+ outs , errs = sub_proc .communicate (timeout = 15 )
210298
299+ if args .exit_on_disposal_received :
300+ for pub_proc in pub_procs :
301+ outs , errs = pub_proc .communicate (timeout = 5 )
211302
212- pub_proc .kill ()
213- ds_proc .kill ()
214- [ds_proc .kill () for ds_proc in ds_procs ]
215- try :
216- sys .exit (os .EX_OK )
217- except AttributeError :
218- sys .exit (0 )
303+ except subprocess .TimeoutExpired :
304+ print ('Target process timed out, terminating...' )
305+ terminate_ok = False
219306
307+ cleanup (pub_procs , sub_procs , ds_procs )
308+ terminate (terminate_ok )
220309
221310if __name__ == '__main__' :
222311
0 commit comments