Commits

Martin Gergov  committed 3def845

Client connects asynchronously and basic IO also works.

  • Participants
  • Parent commits c80c8c8
  • Branches fix_udt_event_handling

Comments (0)

Files changed (5)

 
 #) More options related purely to udt should be included.(e.g. controlling
    socket creation, buffers, timeouts)
+
+#) UDT does not provide separate error events, instead these will be catched
+   on an attempt read or write. This means that connectionLost will NOT be
+   called as soon as the connection is lost for example.
+
+#) It is highly recommended to use the latest revision at the udt-git repo
+   https://sourceforge.net/p/udt/git/ (if it doesn't break
+   compatibility with PyUDT).

File examples/client.py

+from socket import AF_INET, SOCK_DGRAM, AI_PASSIVE
+from udt4 import socket, listen, UDTepoll, bind, recvmsg, UDTException
 from udt4 import startup, cleanup, accept, UDT_EPOLL_IN, UDT_EPOLL_OUT, UDT_EPOLL_ERR
+from udt4 import EASYNCRCV, sendmsg
 from socket import AF_INET, SOCK_DGRAM, AI_PASSIVE
 import udt4
 from udt4.pyudt import UdtSocket
 import gc
+import random
 from time import sleep
 
+ok = True
 startup()
+poller = UDTepoll()
 s = UdtSocket(AF_INET, SOCK_DGRAM, AI_PASSIVE)
+s.setblocking(False)
 print s.UDTSOCKET
-s.setblocking(True)
-s.setsockopt(udt4.UDT_RCVTIMEO, 1000)
-s.connect(("127.0.0.1", 5000))
-s.sendmsg("HELLO!", 7)
+counter = 0
+poller.add_usock(s.UDTSOCKET, UDT_EPOLL_IN)
+
+# try:
+#     while 1:
+#         sock, addr = s.accept()
+#         sleep(20)
+        
+# except KeyboardInterrupt:
+#     exit("Bye!")
+connected = False
 try:
     while True:
+        print "---------------------------------------------"
+        sets = [[], []]
         try:
-            mess = s.recvmsg(1000)
-        except udt4.UDTException as ue:
+            sets = poller.wait(True, True, 1000, False, False)
+        except UDTException as ue:
             if ue[0] == 6003:
                 pass
-            else:
-                raise
-        else:
-            print repr(mess)
-            s.sendmsg(mess)
-            pass
-except udt4.UDTException as ue:
-    print ue
-    exit("Bye!")
+        print sets
+
+        if connected:
+            poller.add_usock(s.UDTSOCKET , UDT_EPOLL_OUT)
+            connected = False
+
+        if ok:
+            try:
+                connectResult = s.connect_ex(("127.0.0.1", 5000))
+            except udt.UDTException as ue:
+                connectResult = ue[0]
+
+            if connectResult == 0:
+                print "CONNECTED!"
+                connected = True
+                ok = False
+
+        for socket in sets[0]:
+            if (socket.UDTSOCKET == s.UDTSOCKET.UDTSOCKET):
+                try:
+                    mess = recvmsg(socket, 100)
+                    print "READ:", socket,";", "RECV:", mess
+                    if counter < 100:
+                        poller.add_usock(socket , UDT_EPOLL_OUT)
+                    #sendmsg(socket.UDTSOCKET, mess, len(mess))
+                except udt4.UDTException as ue:
+                    print "ERR:",ue
+                    if ue[0] == 6002:
+                        poller.remove_usock(socket)
+                        poller.add_usock(socket, UDT_EPOLL_IN)
+                        pass
+            gc.collect()
+
+        for socket in sets[1]:
+            if s.UDTSOCKET.UDTSOCKET == socket.UDTSOCKET:
+                mess = "HELLO!"
+                mess = sendmsg(socket, mess, len(mess))
+                poller.remove_usock(socket)
+                poller.add_usock(socket, UDT_EPOLL_IN)
+
+                counter += 1
+
+            gc.collect()
 except KeyboardInterrupt:
-    pass
-
-print "NEXT"
-s.sendmsg("HELLO!", 7)
-
-try:
-    while True:
-        try:
-            mess = s.recvmsg(1000)
-        except udt4.UDTException as ue:
-            if ue[0] == 6003:
-                print "TIMEOUT"
-                pass
-            else:
-                raise
-        else:
-            print repr(mess)
-            s.sendmsg(mess)
-except udt4.UDTException as ue:
-    print ue
+    poller.remove_usock(s.UDTSOCKET)
+    s.close()
+    cleanup()
     exit("Bye!")
 
-except KeyboardInterrupt:
-    exit("Bye!")
-cleanup()
 
-# import sys
-# sys.path.append("../")
-
-# from udt4twisted import udtepollreactor
-# udtepollreactor.install()
-
-# from twisted.python import log
-# from twisted.internet import reactor, defer
-# from twisted.internet.protocol import DatagramProtocol
-# from twisted.python import log
-
-# log.startLogging(sys.stdout)
-
-
-# class UDTClient(DatagramProtocol):
-
-#     identifier = 0
-
-#     def doSend1(self):
-#         self.transport.write("?>!{0}".format(self.identifier),
-#                              ("127.0.0.1", 5000))
-
-#     def datagramReceived(self, data, (host, port)):
-#         print "Received %r" % data
-#         self.transport.write("bomb!!!{0}".format(self.identifier),
-#                                                 ("127.0.0.1", 5000))
-
-# udtclient = UDTClient()
-# udtclient2 = UDTClient()
-# udtclient2.identifier = 1
-# print udtclient2.identifier
-# reactor.connectUDT("127.0.0.1", 5000, udtclient)
-# reactor.connectUDT("127.0.0.1", 5000, udtclient2)
-# reactor.callLater(1, udtclient.doSend1)
-# reactor.callLater(1, udtclient2.doSend1)
-# reactor.run()
-
-# from udt4 import socket, listen, UDTepoll, bind, recvmsg, UDTException
-# from udt4 import startup, cleanup, accept, UDT_EPOLL_IN, UDT_EPOLL_OUT, UDT_EPOLL_ERR
-# from udt4 import EASYNCRCV, sendmsg
-# from socket import AF_INET, SOCK_DGRAM, AI_PASSIVE
-# import udt4
-# from udt4.pyudt import UdtSocket
-# from time import sleep 
 
 # startup()
 # s = UdtSocket(AF_INET, SOCK_DGRAM, AI_PASSIVE)
 # print s.UDTSOCKET
-# s.setblocking(False)
-# s.connect_ex(("127.0.0.1", 5000))
-# poller = UDTepoll()
-# poller.add_usock(s.UDTSOCKET, UDT_EPOLL_IN)
+# s.setblocking(True)
+# s.setsockopt(udt4.UDT_RCVTIMEO, 1000)
+# s.connect(("127.0.0.1", 5000))
+# s.sendmsg("HELLO!", 7)
+# try:
+#     while True:
+#         try:
+#             mess = s.recvmsg(1000)
+#         except udt4.UDTException as ue:
+#             if ue[0] == 6003:
+#                 pass
+#             else:
+#                 raise
+#         else:
+#             print repr(mess)
+#             s.sendmsg(mess)
+#             pass
+# except udt4.UDTException as ue:
+#     print ue
+#     exit("Bye!")
+# except KeyboardInterrupt:
+#     pass
+
+# print "NEXT"
+# s.sendmsg("HELLO!", 7)
 
 # try:
 #     while True:
-#         sets = poller.wait(True, True, 1000, False, False)
-#         #print sets
-#         for socket in sets[0]:
-#             if (socket.UDTSOCKET == s.UDTSOCKET.UDTSOCKET):
-#                 print 
-#                 try:
-#                     mess = recvmsg(socket.UDTSOCKET, 200)
-#                     print "READ:", socket, ";RECV:", mess
-#                     sendmsg(socket.UDTSOCKET, mess, len(mess))
-#                 except udt4.UDTException as ue:
-#                     print "ERR:",ue
-#                     poller.remove_usock(socket, UDT_EPOLL_IN)
-#                     if ue[0] == 6002:
-#                         poller.add_usock(socket, UDT_EPOLL_IN)
-#             gc.collect()
+#         try:
+#             mess = s.recvmsg(1000)
+#         except udt4.UDTException as ue:
+#             if ue[0] == 6003:
+#                 print "TIMEOUT"
+#                 pass
+#             else:
+#                 raise
+#         else:
+#             print repr(mess)
+#             s.sendmsg(mess)
+# except udt4.UDTException as ue:
+#     print ue
+#     exit("Bye!")
 
 # except KeyboardInterrupt:
 #     exit("Bye!")
-# poller.remove_usock(s.UDTSOCKET)
 # cleanup()

File examples/newclient.py

 
     def dataReceived(self, data):
         log.msg("received {0} from {1}".format(data, self.transport.getPeer()))
-        self.transport.write("bomb!!!")
+        self.transport.write(data)
 
     def connectionLost(self, reason):
         print "CONN LOST!"
 
 class EchoFactory(ClientFactory):
+    conn = []
 
     def startedConnecting(self, connector):
         print 'Started to connect.'
 
     def buildProtocol(self, addr):
         print 'Connected.'
-        return Echo()
+        self.conn.append(Echo())
+        return self.conn[-1]
 
     def clientConnectionLost(self, connector, reason):
         print 'Lost connection.  Reason:', reason
     def connect_to_udt_port(self, res):
         self.echo = EchoFactory()
         self.port = reactor.connectUDT("127.0.0.1", 5000, self.echo)
+        reactor.callLater(3, self.send_after_time, self.echo)
         print self.port
 
+    def send_after_time(self, factory):
+        factory.conn[0].transport.write("HEQQ!")
+                      
     def print_hello(self, res):
         print "Hello!"
         return "hello"
 VERSION = {
         'major' : 0,
         'minor' : 2,
-        'patch' : 3,
+        'patch' : 4,
         }
 # Utility function to read the README file.
 # Used for the long_description.  It's nice, because now 1) we have a top level

File udt4twisted/udt.py

         # in the result.
         return buffer(bObj, offset) + b"".join(bArray)
 
-# class Connector2(Port2):
-
-#     def doConnect(self, fd=None):
-#         if not fd:
-#             self.socket = self.createInternetSocket()
-#             self.fileno = self.socket.fileno
-#             self.socket.connect_ex(self._connectedAddr)
-#             self._connectDone()
-#             return
-
-#         self._read(fd, self._connectedAddr)
-
-#     def doRead(self, fd=None):
-#         why = Port.doRead(self, self.socket.UDTSOCKET, self._connectedAddr)
-#         #FIXME
-#         if why:
-#             if (why[0] == ECONNLOST):
-#                 self.stopListening()
-#         return why
-
 
 @implementer(interfaces.ITCPTransport, interfaces.ISystemHandle)
 class Connection(abstract.FileDescriptor, tcp._AbortingMixin):
             #self._maybePauseProducer()
             #self.startWriting()
             #Really start writable event
-            self.reactor._poller.add_usock(self.socket.UDTSOCKET,
-                                           self.reactor._POLL_OUT)
+            try:
+                self.reactor._poller.add_usock(self.socket.UDTSOCKET,
+                                               self.reactor._POLL_OUT)
+            except udt.UDTException as ue:
+                log.msg("Error while writing:{0}".format(ue))
+                self.abortConnection()
 
 
     _messagesAtOnce = 1024
             self.failIfNotConnected(error.getConnectError((
                         staus)))
             return
+        elif status == udt.UDTSTATUS_CONNECTED:
+            #Check for errors
+            #We're connected, finish the connection
+            del self.doWrite
+            del self.doRead
+            # we first stop and then start, to reset
+            #any references to the old doRead
+            self.startReading()
+            self._connectDone()
+            return
 
         # doConnect gets called twice.  The first time we actually need to
         # start the connection attempt.  The second time we don't really
         # is not /particularly/ detrimental to do so.  This should get
         # cleaned up some day, though.
         try:
-            connectResult = self.socket.connect_ex(self.realAddress)
+            connectResult = self.socket.connect(self.realAddress)
         except udt.UDTException as ue:
             connectResult = ue[0]
 
         #TODO
         if connectResult:
-            #socket is connected
-            if connectResult == udt.ECONNSOCK:
-                #delete references
-                del self.doWrite
-                del self.doRead
-                # we first stop and then start, to reset
-                #any references to the old doRead
-                self.stopReading()
-                self.stopWriting()
-                self._connectDone()
-                return
-
-            elif connectResult in (EASYNCRCV,):
+            if connectResult in (EASYNCRCV,):
+                #We're starting to connect to each other
                 self.startReading()
                 # self.startWriting()
                 return
                 self.failIfNotConnected(error.getConnectError((
                             connectResult, connectResult)))
                 return
-
-
-
-        status = self.socket.getsockopt(udt.UDT_STATE)
-        if status == udt.UDTSTATUS_CONNECTED:
-            # If I have reached this point without raising
-            #or returning, that means
-            # that the socket is connected.
-            del self.doWrite
-            del self.doRead
-            # we first stop and then start, to reset
-            #any references to the old doRead
-            self.startReading()
-            self._connectDone()
         else:
-            self.startReading()
-            self.startWriting()
+            #FIXME but it works for now
+            opt = self.socket.getsockopt(udt.UDT_STATE)
+            if opt == udt.UDTSTATUS_CONNECTING:
+                #add monitoring to the reactor
+                self.startReading()
+                self.startWriting()
+                self.reactor._poller.add_usock(self.socket.UDTSOCKET,
+                                               self.reactor._POLL_OUT)
+            elif opt == udt.UDTSTATUS_CONNECTED:
+                #Check for errors
+                #We're connected, finish the connection
+                del self.doWrite
+                del self.doRead
+                # we first stop and then start, to reset
+                #any references to the old doRead
+                self.startReading()
+                self._connectDone()
+                return
 
     def _connectDone(self):
         """