appPublic/sshx.py

@@ -1,6 +1,5 @@
 import os
 import sys
-import time
 import shlex
 from functools import partial
 from threading import Thread
@@ -19,7 +18,6 @@ class SSHNode:
 				"password":password,
 				"port":port
 		}
-		print(self.server2)
 		self.jumpers = jumpers
 		self.conn = None
 		self.jumper_conns = []
@@ -111,54 +109,48 @@ class SSHNode:
 										term_size=(80,24),
 										encoding='utf-8'
 		)
+		buffer = ''
 		
+		msglen = len(xmsgs)
+		for i in range(msglen):
+			if xmsgs[i][0]:
+				break
+			s = xmsgs[i][1].format(**ns)
+			proc.stdin.write(s)
 		keyin = False
 		def feed_data(xmsgs, debug_input):
 			if len(xmsgs) == 0:
-				print('#####++#####  xmsgs has zero elements')
 				return
+			a = xmsgs[0]
+			xmsgs.reverse()
+			xmsgs.pop()
+			xmsgs.reverse()
+			s = a[1].format(**ns)
+			proc.stdin.write(s)
+			if debug_input:
+				debug_input(f'{s=},{a=}')
 			keyin = True
-			a = xmsgs.pop(0)
-			while True:
-				if a[1] is None:
-					proc.stdin.write_eof()
-					self.running = False
-				else:
-					s = a[1].format(**ns)
-					proc.stdin.write(s)
-
-				if len(xmsgs) == 0 or xmsgs[0][0]:
-					break
-				a = xmsgs.pop(0)
 
 		already_output = False
 		callee = None
 		loop = asyncio.get_event_loop()
-		self.running = True
+		while True:
-		while self.running:
+			if proc.stdout.at_eof():
+				break
+
 			if keyin:
 				keyin = False
 				await proc.stdin.drain()
-			if proc.stdout.at_eof():
-				break
-			tup = proc.collect_output()
-			x = tup[0]
-			if x!='' and show_stdout:
-				if x is None:
-					break
-				if callee:
-					callee.cancel()
-				callee = None
-				show_stdout(x)
-			else:
-				if callee is None:
-					if len(xmsgs) > 0:
-						f = partial(feed_data, xmsgs, show_input)
-						t = xmsgs[0][0] or 0
-						callee = loop.call_later(t, f)
-			await asyncio.sleep(0.05)
 
-		print('##########fininshed##########')
+			if len(xmsgs) > 0 and already_output:
+				f = partial(feed_data, xmsgs, show_input)
+				callee = loop.call_later(xmsgs[0][0], f)
+			x = await proc.stdout.read(1024)
+			if callee:
+				callee.cancel()
+			if show_stdout:
+				show_stdout(x)
+			already_output = True
 
 	async def _run(self, cmd, input=None, stdin=None, stdout=None):
 		if cmd.startswith('l2r'):

appPublic/worker.py

@@ -1,27 +1,21 @@
 import random
 import asyncio
-import inspect
 from functools import wraps
 
-def to_func(func):
+def asyncCall(func):
 	@wraps(func)
 	def wraped_func(*args,**kw):
-		if inspect.iscoroutinefunction(func):
+		task =  asyncio.ensure_future(func(*args,**kw))
-			task =  asyncio.ensure_future(func(*args,**kw))
+		asyncio.gather(task)
-			ret = asyncio.gather(task)
-			return ret
-		return func(*args, **kw)
 	return wraped_func
 
-class AsyncWorker:
+class Worker:
 	def __init__(self,maxtask=50):
 		self.semaphore = asyncio.Semaphore(maxtask)
 
 	async def __call__(self,callee,*args,**kw):
 		async with self.semaphore:
-			if inspect.iscoroutinefunction(callee):
+			return await callee(*args,**kw)
-				return await callee(*args,**kw)
-			return callee(*args, **kw)
 
 	async def run(self,cmd):
 		async with self.semaphore:
@@ -33,16 +27,16 @@ class AsyncWorker:
 			return stdout, stderr
 
 if __name__ == '__main__':
-	def hello(cnt,greeting):
+	async def hello(cnt,greeting):
 		t = random.randint(1,10)
-		# await asyncio.sleep(t)
+		await asyncio.sleep(t)
 		print(cnt,'cost ',t,'seconds to',greeting)
 
+	
 	async def run():
-		w = AsyncWorker()
+		w = Worker()
-		g = [ asyncio.create_task(w(hello,i,'hello world')) for i in range(1000) ]
+		tasks = [ w(hello,i,'hello world') for i in range(1000) ]
-		await asyncio.wait(g)
+		await asyncio.wait(tasks)
-		print('aaaaaaaaaaaaaaaaaaa')
 
 	loop = asyncio.get_event_loop()
 	loop.run_until_complete(run())