[Twisted-Python] How do I get data from a Queue (or "outside") into a Telnet Server?

Mark Bailey mebly5343 at gmail.com
Tue Feb 2 21:47:56 EST 2010

On Feb 2, 2010 8:54 AM, <exarkun at twistedmatrix.com> wrote:

On 02:15 am, mebly5343 at gmail.com wrote:
>Good day, everyone:
>>Good day, everyone:
>I'm trying to learn Python and Twisted at the same time and having fun
>I'm writing an application that is collecting data from multiple
>sources, filtering the data, and providing it to users through a Telnet
>I can set up a polling loop for the server but I can't figure out how to
>send the data to users connected to my server.
It's possible this FAQ entry will help:



Thanks, Jean-Paul.  The following appears to do exactly what I want, based
on your FAQ entry.  It reads data from a socket
connection and sends it to clients connected to the Telnet server.  I think
I've captured the correct instances this time.  :-)

Now, I need to work oncollecting the data and play with tksupport to provide
a GUI...!

Thanks again.


from twisted.conch.telnet import StatefulTelnetProtocol
from twisted.internet import reactor, protocol
from twisted.protocols.basic import LineReceiver

class mySocket(LineReceiver):
    def connectionMade(self):
        self.factory.connection = self
        self.factory.data = []

    def lineReceived(self, line):

    def connectionLost(self, reason):
        self.factory.connection = None

class mySocketFactory(protocol.Factory):
    protocol = mySocket
    def __init__(self):
        self.connection = None
        self.data = []

class TelnetEcho(StatefulTelnetProtocol):
    def connectionMade(self):
        self.factory.connection = self
        self.sendLine("Welcome to the ClusterMerge 0.9Beta 20100203 telnet
port\r\n\narc >\r")

    def lineReceived(self, data):
        data = data.rstrip('\n\r')

        if data.upper() == 'BYE':
            self.factory.connection = None
            self.sendLine("Unrecognized command: %r\r" % (data,))  #  will
implement help, set/filter, and show/filter later

    def connectionLost(self, reason):
        self.factory.connection = None

class TelnetEchoFactory(protocol.Factory):
    protocol = TelnetEcho
    def __init__(self):
        self.connection = None

def checkforspots(telnetinstance, socketinstance):
    while len(socketinstance.data) > 0:
        line = socketinstance.data.pop()
        if telnetinstance.connection:
            telnetinstance.connection.sendLine(line + "\r")

    reactor.callLater(1.0,checkforspots, telnetinstance, socketinstance)

def createTelnetServer(port=7300, myport=7301):
    telnetinstance = TelnetEchoFactory()

    socketinstance = mySocketFactory()
    reactor.listenTCP(myport, socketinstance)

    checkforspots(telnetinstance, socketinstance)

if __name__ == "__main__":
    port = 8023
    reactor.callWhenRunning(createTelnetServer, port)

Twisted-Python mailing list
Twisted-Python at twistedmatrix.com
-------------- next part --------------
An HTML attachment was scrubbed...
URL: http://twistedmatrix.com/pipermail/twisted-python/attachments/20100202/6cadc86a/attachment-0001.htm 

More information about the Twisted-Python mailing list