Merge branch 'master' of git.kaiyuancloud.cn:yumoqing/apppublic

appPublic/sshx.py

@@ -1,5 +1,6 @@
 import os
 import sys
+import time
 import shlex
 from functools import partial
 from threading import Thread
@@ -18,6 +19,7 @@ class SSHNode:
 				"password":password,
 				"port":port
 		}
+		print(self.server2)
 		self.jumpers = jumpers
 		self.conn = None
 		self.jumper_conns = []
@@ -109,48 +111,54 @@ 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()
+			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 debug_input:
-				debug_input(f'{s=},{a=}')
-			keyin = True
+
+				if len(xmsgs) == 0 or xmsgs[0][0]:
+					break
+				a = xmsgs.pop(0)
 
 		already_output = False
 		callee = None
 		loop = asyncio.get_event_loop()
-		while True:
-			if proc.stdout.at_eof():
-				break
-
+		self.running = True
+		while self.running:
 			if keyin:
 				keyin = False
 				await proc.stdin.drain()
-
-			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 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()
-			if show_stdout:
+				callee = None
 				show_stdout(x)
-			already_output = True
+			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##########')
 
 	async def _run(self, cmd, input=None, stdin=None, stdout=None):
 		if cmd.startswith('l2r'):

appPublic/worker.py

@@ -1,21 +1,27 @@
 import random
 import asyncio
+import inspect
 from functools import wraps
 
-def asyncCall(func):
+def to_func(func):
 	@wraps(func)
 	def wraped_func(*args,**kw):
+		if inspect.iscoroutinefunction(func):
 			task =  asyncio.ensure_future(func(*args,**kw))
-		asyncio.gather(task)
+			ret = asyncio.gather(task)
+			return ret
+		return func(*args, **kw)
 	return wraped_func
 
-class Worker:
+class AsyncWorker:
 	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 callee(*args, **kw)
 
 	async def run(self,cmd):
 		async with self.semaphore:
@@ -27,16 +33,16 @@ class Worker:
 			return stdout, stderr
 
 if __name__ == '__main__':
-	async def hello(cnt,greeting):
+	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 = Worker()
-		tasks = [ w(hello,i,'hello world') for i in range(1000) ]
-		await asyncio.wait(tasks)
+		w = AsyncWorker()
+		g = [ asyncio.create_task(w(hello,i,'hello world')) for i in range(1000) ]
+		await asyncio.wait(g)
+		print('aaaaaaaaaaaaaaaaaaa')
 
 	loop = asyncio.get_event_loop()
 	loop.run_until_complete(run())