summaryrefslogtreecommitdiffstats
path: root/examples/telnet/chat-app.py
blob: 1880a362b91cbf1dbe8956f1a8fb378ab15b071d (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
#!/usr/bin/env python
"""
A simple chat application over telnet.
Everyone that connects is asked for his name, and then people can chat with
each other.
"""
import logging
import random

from prompt_toolkit.contrib.telnet.server import TelnetServer
from prompt_toolkit.eventloop import get_event_loop
from prompt_toolkit.formatted_text import HTML
from prompt_toolkit.shortcuts import PromptSession, clear, prompt

# Set up logging
logging.basicConfig()
logging.getLogger().setLevel(logging.INFO)

# List of connections.
_connections = []
_connection_to_color = {}


COLORS = [
    "ansired",
    "ansigreen",
    "ansiyellow",
    "ansiblue",
    "ansifuchsia",
    "ansiturquoise",
    "ansilightgray",
    "ansidarkgray",
    "ansidarkred",
    "ansidarkgreen",
    "ansibrown",
    "ansidarkblue",
    "ansipurple",
    "ansiteal",
]


async def interact(connection):
    write = connection.send
    prompt_session = PromptSession()

    # When a client is connected, erase the screen from the client and say
    # Hello.
    clear()
    write("Welcome to our chat application!\n")
    write("All connected clients will receive what you say.\n")

    name = await prompt_session.prompt_async(message="Type your name: ")

    # Random color.
    color = random.choice(COLORS)
    _connection_to_color[connection] = color

    # Send 'connected' message.
    _send_to_everyone(connection, name, "(connected)", color)

    # Prompt.
    prompt_msg = HTML('<reverse fg="{}">[{}]</reverse> &gt; ').format(color, name)

    _connections.append(connection)
    try:
        # Set Application.
        while True:
            try:
                result = await prompt_session.prompt_async(message=prompt_msg)
                _send_to_everyone(connection, name, result, color)
            except KeyboardInterrupt:
                pass
    except EOFError:
        _send_to_everyone(connection, name, "(leaving)", color)
    finally:
        _connections.remove(connection)


def _send_to_everyone(sender_connection, name, message, color):
    """
    Send a message to all the clients.
    """
    for c in _connections:
        if c != sender_connection:
            c.send_above_prompt(
                [
                    ("fg:" + color, "[%s]" % name),
                    ("", " "),
                    ("fg:" + color, "%s\n" % message),
                ]
            )


def main():
    server = TelnetServer(interact=interact, port=2323)
    server.start()
    get_event_loop().run_forever()


if __name__ == "__main__":
    main()