Mercurial > libervia-backend
annotate sat_frontends/jp/cmd_pipe.py @ 3913:944f51f9c2b4
core (xmpp): make `send` a blocking method, fix `sendMessageData` calls:
original `send` method is blocking, and it is used as such by Wokkel and thus can't be
changed to an async method easily. However, an Async method is necessary to have an async
trigger at the very end of the send workflow for end-to-end encryption. To workaround
that, `send` is an async method which call `a_send`, an async method which actually does
the sending. This way legacy method can still call `send` while `a_send` can be await
otherwise.
Fix calls to `sendMessageData`: the method now being an `async` one, `ensureDeferred` had
to be used in some calls.
| author | Goffi <goffi@goffi.org> |
|---|---|
| date | Sat, 24 Sep 2022 16:31:39 +0200 |
| parents | be6d91572633 |
| children | 524856bd7b19 |
| rev | line source |
|---|---|
| 3137 | 1 #!/usr/bin/env python3 |
| 2 | |
| 815 | 3 |
| 4 # jp: a SAT command line tool | |
| 3479 | 5 # Copyright (C) 2009-2021 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 ) |