mirror of https://github.com/jesopo/irctokens
add a stateful encoder (irctokens.StatefulEncoder())
This commit is contained in:
parent
3cdc163b87
commit
49d2e78475
20
README.md
20
README.md
|
@ -31,17 +31,30 @@ if line.command == "PRIVMSG":
|
||||||
```
|
```
|
||||||
|
|
||||||
### stateful
|
### stateful
|
||||||
|
|
||||||
|
below is an example of a fully socket-wise safe IRC client connection that will
|
||||||
|
connect and join a channel. both protocol sending and receiving are handled by
|
||||||
|
irctokens.
|
||||||
|
|
||||||
```python
|
```python
|
||||||
|
|
||||||
import irctokens, socket
|
import irctokens, socket
|
||||||
|
|
||||||
|
NICK = "nickname"
|
||||||
|
CHAN = "#channel"
|
||||||
|
|
||||||
d = irctokens.StatefulDecoder()
|
d = irctokens.StatefulDecoder()
|
||||||
|
e = irctokens.StatefulEncoder()
|
||||||
s = socket.socket()
|
s = socket.socket()
|
||||||
s.connect(("127.0.0.1", 6667))
|
s.connect(("127.0.0.1", 6667))
|
||||||
|
|
||||||
def _send(line):
|
def _send(line):
|
||||||
s.send(f"{line}\r\n".encode("utf8"))
|
print(f"> {line.format()}")
|
||||||
|
e.push(line)
|
||||||
|
while e.pending():
|
||||||
|
e.pop(s.send(e.pending()))
|
||||||
|
|
||||||
_send(irctokens.format("USER", ["username", "0", "*", "real name"]))
|
_send(irctokens.format("USER", [NICK, "0", "*", "real name"]))
|
||||||
_send(irctokens.format("NICK", ["nickname"]))
|
_send(irctokens.format("NICK", ["nickname"]))
|
||||||
|
|
||||||
while True:
|
while True:
|
||||||
|
@ -51,11 +64,12 @@ while True:
|
||||||
break
|
break
|
||||||
|
|
||||||
for line in lines:
|
for line in lines:
|
||||||
|
print(f"< {line.format()}")
|
||||||
if line.command == "PING":
|
if line.command == "PING":
|
||||||
to_send = irctokens.format("PONG", [line.params[0]])
|
to_send = irctokens.format("PONG", [line.params[0]])
|
||||||
_send(to_send)
|
_send(to_send)
|
||||||
|
|
||||||
elif line.command == "001":
|
elif line.command == "001":
|
||||||
to_send = irctokens.format("JOIN", ["#test"])
|
to_send = irctokens.format("JOIN", [CHAN])
|
||||||
_send(to_send)
|
_send(to_send)
|
||||||
```
|
```
|
||||||
|
|
|
@ -1,2 +1,2 @@
|
||||||
from .protocol import Line, tokenise, format
|
from .protocol import Line, tokenise, format
|
||||||
from .stateful import StatefulDecoder
|
from .stateful import StatefulDecoder, StatefulEncoder
|
||||||
|
|
|
@ -21,3 +21,16 @@ class StatefulDecoder(object):
|
||||||
except UnicodeDecodeError as e:
|
except UnicodeDecodeError as e:
|
||||||
decode_lines.append(line.decode(self._fallback))
|
decode_lines.append(line.decode(self._fallback))
|
||||||
return [tokenise(l) for l in decode_lines]
|
return [tokenise(l) for l in decode_lines]
|
||||||
|
|
||||||
|
class StatefulEncoder(object):
|
||||||
|
def __init__(self):
|
||||||
|
self._buffer = b""
|
||||||
|
|
||||||
|
def pending(self) -> bytes:
|
||||||
|
return self._buffer
|
||||||
|
|
||||||
|
def push(self, line: Line) -> bytes:
|
||||||
|
self._buffer += f"{line.format()}\r\n".encode("utf8")
|
||||||
|
return self._buffer
|
||||||
|
def pop(self, byte_count: int):
|
||||||
|
self._buffer = self._buffer[byte_count:]
|
||||||
|
|
|
@ -0,0 +1,17 @@
|
||||||
|
import unittest
|
||||||
|
import irctokens
|
||||||
|
|
||||||
|
class TestPush(unittest.TestCase):
|
||||||
|
def test(self):
|
||||||
|
e = irctokens.StatefulEncoder()
|
||||||
|
line = irctokens.tokenise("PRIVMSG #channel hello")
|
||||||
|
e.push(line)
|
||||||
|
self.assertEqual(e.pending(), b"PRIVMSG #channel hello\r\n")
|
||||||
|
|
||||||
|
class TestPop(unittest.TestCase):
|
||||||
|
def test(self):
|
||||||
|
e = irctokens.StatefulEncoder()
|
||||||
|
line = irctokens.tokenise("PRIVMSG #channel hello")
|
||||||
|
e.push(line)
|
||||||
|
e.pop(len(b"PRIVMSG #channel hello"))
|
||||||
|
self.assertEqual(e.pending(), b"\r\n")
|
Loading…
Reference in New Issue