Mercurial > libervia-backend
annotate sat_frontends/jp/cmd_pipe.py @ 3040:fee60f17ebac
jp: jp asyncio port:
/!\ this commit is huge. Jp is temporarily not working with `dbus` bridge /!\
This patch implements the port of jp to asyncio, so it is now correctly using the bridge
asynchronously, and it can be used with bridges like `pb`. This also simplify the code,
notably for things which were previously implemented with many callbacks (like pagination
with RSM).
During the process, some behaviours have been modified/fixed, in jp and backends, check
diff for details.
| author | Goffi <goffi@goffi.org> |
|---|---|
| date | Wed, 25 Sep 2019 08:56:41 +0200 |
| parents | ab2696e34d29 |
| children | 9d0df638c8b4 |
| rev | line source |
|---|---|
| 1960 | 1 #!/usr/bin/env python2 |
| 815 | 2 # -*- coding: utf-8 -*- |
| 3 | |
| 4 # jp: a SAT command line tool | |
| 2771 | 5 # Copyright (C) 2009-2019 Jérôme Poisson (goffi@goffi.org) |
| 815 | 6 |
| 7 # This program is free software: you can redistribute it and/or modify | |
| 8 # it under the terms of the GNU Affero General Public License as published by | |
| 9 # the Free Software Foundation, either version 3 of the License, or | |
| 10 # (at your option) any later version. | |
| 11 | |
| 12 # This program is distributed in the hope that it will be useful, | |
| 13 # but WITHOUT ANY WARRANTY; without even the implied warranty of | |
| 14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the | |
| 15 # GNU Affero General Public License for more details. | |
| 16 | |
| 17 # You should have received a copy of the GNU Affero General Public License | |
| 18 # along with this program. If not, see <http://www.gnu.org/licenses/>. | |
| 19 | |
| 3040 | 20 import socket |
| 21 import asyncio | |
| 22 import errno | |
| 23 from functools import partial | |
| 817 | 24 from sat_frontends.jp import base |
|
2489
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
25 from sat_frontends.jp.constants import Const as C |
| 3040 | 26 from sat_frontends.jp import xmlui_manager |
| 817 | 27 import sys |
| 814 | 28 from sat.core.i18n import _ |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
29 from sat_frontends.tools import jid |
| 0 | 30 |
| 817 | 31 __commands__ = ["Pipe"] |
| 0 | 32 |
|
2489
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
33 START_PORT = 9999 |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
34 |
|
2624
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
35 |
| 817 | 36 class PipeOut(base.CommandBase): |
| 37 def __init__(self, host): | |
|
2624
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
38 super(PipeOut, self).__init__(host, "out", help=_("send a pipe a stream")) |
| 393 | 39 |
| 817 | 40 def add_parser_options(self): |
|
2624
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
41 self.parser.add_argument( |
| 3028 | 42 "jid", help=_("the destination jid") |
|
2624
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
43 ) |
|
401
b2caa2615c4c
jp roster name manegement + Pipe transfer
Goffi <goffi@goffi.org>
parents:
393
diff
changeset
|
44 |
| 3040 | 45 async def start(self): |
| 46 """ Create named pipe, and send stdin to it """ | |
| 47 try: | |
| 48 port = await self.host.bridge.streamOut( | |
| 49 await self.host.get_full_jid(self.args.jid), | |
| 50 self.profile, | |
| 51 ) | |
| 52 except Exception as e: | |
| 53 self.disp(f"can't start stream: {e}", error=True) | |
| 54 self.host.quit(C.EXIT_BRIDGE_ERRBACK) | |
| 55 else: | |
| 56 # FIXME: we use temporarily blocking code here, as it simplify | |
| 57 # asyncio port: "loop.connect_read_pipe(lambda: reader_protocol, | |
| 58 # sys.stdin.buffer)" doesn't work properly when a file is piped in | |
| 59 # (we get a "ValueError: Pipe transport is for pipes/sockets only.") | |
| 60 # while it's working well for simple text sending. | |
|
2489
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
61 |
| 3040 | 62 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| 63 s.connect(("127.0.0.1", int(port))) | |
| 64 | |
| 65 while True: | |
| 66 buf = sys.stdin.buffer.read(4096) | |
| 67 if not buf: | |
| 68 break | |
| 69 try: | |
| 70 s.sendall(buf) | |
| 71 except socket.error as e: | |
| 72 if e.errno == errno.EPIPE: | |
| 73 sys.stderr.write(f"e\n") | |
| 74 self.host.quit(1) | |
| 75 else: | |
| 76 raise e | |
| 77 self.host.quit() | |
|
2489
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
78 |
|
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
79 |
| 3040 | 80 async def handle_stream_in(reader, writer, host): |
| 81 """Write all received data to stdout""" | |
| 82 while True: | |
| 83 data = await reader.read(4096) | |
| 84 if not data: | |
| 85 break | |
| 86 sys.stdout.buffer.write(data) | |
| 87 try: | |
| 88 sys.stdout.flush() | |
| 89 except IOError as e: | |
| 90 sys.stderr.write(f"{e}\n") | |
| 91 break | |
| 92 host.quitFromSignal() | |
|
401
b2caa2615c4c
jp roster name manegement + Pipe transfer
Goffi <goffi@goffi.org>
parents:
393
diff
changeset
|
93 |
| 817 | 94 |
| 95 class PipeIn(base.CommandAnswering): | |
| 96 def __init__(self, host): | |
|
2624
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
97 super(PipeIn, self).__init__(host, "in", help=_("receive a pipe stream")) |
|
2489
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
98 self.action_callbacks = {"STREAM": self.onStreamAction} |
|
587
952322b1d490
Remove trailing whitespaces.
Emmanuel Gil Peyrot <linkmauve@linkmauve.fr>
parents:
572
diff
changeset
|
99 |
| 817 | 100 def add_parser_options(self): |
|
2624
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
101 self.parser.add_argument( |
|
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
102 "jids", |
|
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
103 nargs="*", |
|
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
104 help=_('Jids accepted (none means "accept everything")'), |
|
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
105 ) |
| 0 | 106 |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
107 def getXmluiId(self, action_data): |
|
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
108 try: |
|
2624
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
109 xml_ui = action_data["xmlui"] |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
110 except KeyError: |
| 3028 | 111 self.disp(_("Action has no XMLUI"), 1) |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
112 else: |
| 3040 | 113 ui = xmlui_manager.create(self.host, xml_ui) |
| 114 if not ui.submit_id: | |
| 3028 | 115 self.disp(_("Invalid XMLUI received"), error=True) |
| 3040 | 116 self.quitFromSignal(C.EXIT_INTERNAL_ERROR) |
| 117 return ui.submit_id | |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
118 |
| 3040 | 119 async def onStreamAction(self, action_data, action_id, security_limit, profile): |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
120 xmlui_id = self.getXmluiId(action_data) |
|
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
121 if xmlui_id is None: |
| 3040 | 122 self.host.quitFromSignal(C.EXIT_ERROR) |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
123 try: |
|
2624
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
124 from_jid = jid.JID(action_data["meta_from_jid"]) |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
125 except KeyError: |
| 3040 | 126 self.disp(_("Ignoring action without from_jid data"), error=True) |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
127 return |
|
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
128 |
|
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
129 if not self.bare_jids or from_jid.bare in self.bare_jids: |
|
2489
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
130 host, port = "localhost", START_PORT |
|
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
131 while True: |
|
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
132 try: |
| 3040 | 133 server = await asyncio.start_server( |
| 134 partial(handle_stream_in, host=self.host), host, port) | |
|
2489
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
135 except socket.error as e: |
|
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
136 if e.errno == errno.EADDRINUSE: |
|
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
137 port += 1 |
|
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
138 else: |
|
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
139 raise e |
|
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
140 else: |
|
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
141 break |
| 3028 | 142 xmlui_data = {"answer": C.BOOL_TRUE, "port": str(port)} |
| 3040 | 143 await self.host.bridge.launchAction( |
| 144 xmlui_id, xmlui_data, profile_key=profile) | |
| 145 async with server: | |
| 146 await server.serve_forever() | |
|
2489
e2a7bb875957
plugin pipe/stream, file transfert: refactoring and improvments:
Goffi <goffi@goffi.org>
parents:
2483
diff
changeset
|
147 self.host.quitFromSignal() |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
148 |
| 3040 | 149 async def start(self): |
|
1670
3690b4d4157e
jp (pipe): pipe commands now use the new CommandAnswering API (with actionNew)
Goffi <goffi@goffi.org>
parents:
1396
diff
changeset
|
150 self.bare_jids = [jid.JID(jid_).bare for jid_ in self.args.jids] |
| 3040 | 151 await self.start_answering() |
| 817 | 152 |
| 153 | |
| 154 class Pipe(base.CommandBase): | |
| 155 subcommands = (PipeOut, PipeIn) | |
| 156 | |
| 157 def __init__(self, host): | |
|
2624
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
158 super(Pipe, self).__init__( |
|
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
159 host, "pipe", use_profile=False, help=_("stream piping through XMPP") |
|
56f94936df1e
code style reformatting using black
Goffi <goffi@goffi.org>
parents:
2562
diff
changeset
|
160 ) |