Многопоточное сокетное программирование в python

Question

Я работаю над программированием сокетов клиент / сервер с использованием Python. Я мог бы установить его без каких-либо проблем, но с потоками, я потерян. Я понятия не имею, как выполнить эту часть. Я также посмотрел документацию по Python, но не смог найти то, что ожидал.

Я пытаюсь добиться того, чтобы сервер принимал соединение от клиента, а также отслеживал каталог для создания любого файла. Я пытаюсь запустить их обоих параллельно, но это не сработало.

#!/usr/bin/python           # This is server.py file

import socket               # Import socket module
import sys,os
import time
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

class MyHandler(PatternMatchingEventHandler):
    patterns = ["*.challenger", "*.challenger"]

    def process(self, event):
        """
        event.event_type
            'modified' | 'created' | 'moved' | 'deleted'
        event.is_directory
            True
        event.src_path
            /home/abcd/Maildir/new/
        """
        # the file will be processed there
        print event.src_path, event.event_type  # print now only for degug

    def on_modified(self, event):
        self.process(event)
        flag = '1'
        print flag

    def on_created(self, event):
        self.process(event)

    def on_any_event(self,event):
        if event.event_type == 'created':
            send()


s = socket.socket()         # Create a socket object
host = socket.gethostname() # Get local machine name
port = 12346                # Reserve a port for your service.
s.bind((host, port))        # Bind to the port
flag = '0'
s.listen(5)                 # Now wait for client connection.
while True:
    c, addr = s.accept()     # Establish connection with client.
    print 'Got connection from', addr

#   c.send('Thank you for connecting')
#   c.send(flag)        

    if __name__ == '__main__':

       args = sys.argv[1:]
       observer = Observer()
       observer.schedule(MyHandler(), path='/home/abcd/Maildir/new')
       observer.start()

    try:
        while True:
          time.sleep(1)

    except KeyboardInterrupt:
         observer.stop()

observer.join()


c.send(flag)
c.close()                # Close the connection

Код, который я разместил, без потоков, потому что с потоком он вообще не работал.

Я использую watchdog для мониторинга /home/abcd/Maildir/new для мониторинга любого нового созданного файла электронной почты, и если он создается, я хочу отправить подтверждение (в данном случае флаг) клиенту, который имеет электронную почту был получен. Сокетное соединение и Watchdog работали хорошо по отдельности, но я не знаю, почему они не будут работать вместе?:(

Как я могу поместить эти методы в разные потоки, чтобы они работали параллельно? Спасибо за вашу помощь.

РЕДАКТИРОВАТЬ: код после ввода Крейга:

import threading
import socket               # Import socket module
import sys,os
import time
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

def send_message(conn, flag):
    conn.send(flag)


class MyHandler(PatternMatchingEventHandler):
    patterns = ["*.challenger", "*.challenger"]

    def __init__(self, conn, *args, **kwargs):
        super(MyHandler, self).__init__(*args, **kwargs)
        self.conn_list = conn

    def process(self, event):
        """
        event.event_type
            'modified' | 'created' | 'moved' | 'deleted'
        event.is_directory
            True
        event.src_path
            /home/abcd/Maildir/new/
        """
        print "I reached here too"
        time.sleep(5)
        print event.src_path, event.event_type  # print now only for degug

    def on_created(self, event):
        flag = '1'
        threads = [threading.Thread(target=send_message, args=(conn, flag)) for conn in self.conn_list]
        for t in threads:
            t.start()
        print "on_created"
        self.process(event)
        for t in threads:
            t.join()


from collections import deque
conn_list = deque()

if __name__ == '__main__':
    s = socket.socket()         # Create a socket object
    host = socket.gethostname()  # Get local machine name
    port = 12346                # Reserve a port for your service.
    s.bind((host, port))        # Bind to the port
    flag = '0'
    s.listen(5)                 # Now wait for client connection.
    observer = Observer()
    observer.schedule(MyHandler(conn_list), path='/home/abcd/Maildir/new/')
    observer.start()
    print "Before True"
while True:
     try:
        c, addr = s.accept()     # Establish connection with client.
        print 'Got connection from', addr
        conn_list.append(c)
        time.sleep(1)
        print "I started Observer"
     except KeyboardInterrupt:
        observer.stop()
        while conn_list:
            conn_list.pop().close()
        print "Connections closed"
        break

observer.join()

РЕДАКТИРОВАТЬ 2:

Когда я отправляю электронное письмо от клиента к серверу, я получаю правильный результат в первый раз, но когда я подключаюсь еще раз, вывод получается странным ("In message" печатается дважды, значение флага также печатается дважды, и я получаю сообщение об ошибке сломанного канала)

Модифицированный Server.py после ввода Крейга:

import threading
import socket               # Import socket module
import sys,os
import time
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler


def send_message(conn, flag):
    print "In message"
    conn.send(flag)
    print "flag"+flag

class MyHandler(PatternMatchingEventHandler):
    patterns = ["*.challenger", "*.challenger"]

    def __init__(self, conn, *args, **kwargs):
        super(MyHandler, self).__init__(*args, **kwargs)
        self.conn_list = conn

    def process(self, event):
        time.sleep(5)
        print "In process"
        print event.src_path, event.event_type  # print now only for degug

    def on_created(self, event):
        flag = '1'
        print "before process event"
        self.process(event)
        print "after process, before thread target"
        threads = [threading.Thread(target=send_message, args=(conn, flag)) for$
        flag = '0'
        for t in threads:
            t.start()
#        print "Before process(Event)"
#        self.process(event)
        print "after process event"
        for t in threads:
            t.join()

s = socket.socket()         # Create a socket object
host = socket.gethostname()  # Get local machine name
port = 12345                # Reserve a port for your service.
s.bind((host, port))        # Bind to the port
flag = '0'
s.listen(5)                 # Now wait for client connection.
print "flag before: "+flag

from collections import deque
conn_list = deque()

if __name__ == '__main__':
    observer = Observer()
    observer.start()
    args = sys.argv[1:]
    observer.schedule(MyHandler(conn_list), path='/home/abcd/Maildir/new')
while True:
    try:
        c, addr = s.accept()     # Establish connection with client.
        print 'Got connection from', addr
        flag = '0'
        print flag
        conn_list.append(c)

    except KeyboardInterrupt:
        observer.stop()
        while conn_list:
            conn_list.pop().close()
        print "Connections closed"
        break

observer.join()

Вот вывод:

Got connection from ('72.123.27.223', 39844)
0
before process event
In process
/home/abcd/Maildir/new/1425403821.V801I2ac232cM275759.challenger created
after process, before thread target
In message
In message
flag1In message
 after process event

 flag1
Exception in thread Thread-7:
Traceback (most recent call last):
  File "/usr/lib/python2.7/threading.py", line 810, in __bootstrap_inner
    self.run()
  File "/usr/lib/python2.7/threading.py", line 763, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/home/paras/server42.py", line 11, in send_message
    conn.send(flag)
error: [Errno 32] Broken pipe

Вот мой client.py - все, что я пытаюсь сделать, это проверить задержку электронной почты от клиента к серверу и получить ответ от сервера.

#!/usr/bin/python           # This is client.py file
import sys,smtplib
import socket               # Import socket module
import threading,time
import urllib
import random
import datetime
#class myThread(threading.Thread):
#    def __init__ (self,name):
#        threading.Thread.__init__(self)
#   self.name=name

#    def run(self):
#        connection()    
#        print "Starting"+self.name

def connection():
    s = socket.socket()         # Create a socket object
    host = socket.gethostbyname('server.edu') # Get local machine name
    port = 12345                # Reserve a port for your service.
    try:
        s.connect((host, port))
#        print s.recv(1024)
        t1 = datetime.datetime.now()
        sendmail()
    t2 = datetime.datetime.now()
        print str(t1),str(t2) 
        print "Time taken to send e-mail from client to server: "+str(t2-t1)
        print "came back"
        flag = s.recv(1024)
        print flag
        if (flag=='1'):
            t3=datetime.datetime.now()   
        print "Time taken to receive response from server: "+str(t3-t2)
        s.close                     # Close the socket when done
    except KeyboardInterrupt:
        s.close

def sendmail():
    fromaddr = '[e-mail]'

    toaddrs = '[e-mail]'
    url = "http://www.google.com"
    seq = str(random.randint(1,9))
    msg = 'URL: '+ url  + '\n'+'SEQ:'+seq+'\n'
    print "In sendmail"
#print msg

# Credentials (if needed)
    username = '[e-mail]'
    password = 'somepassword'
#The actual mail send
    server = smtplib.SMTP('smtp.gmail.com:587')
#server = smtplib.SMTP('localhost')
    server.starttls()
    server.login(username,password)
    server.sendmail(fromaddr, toaddrs, msg)
    server.quit()
    print "email sent"
    return

if __name__ == "__main__":
    connection()

Пожалуйста, помогите мне. Благодарю.

Answer 1

Это не совсем то, что вы просили, так как сервер не отправляет данные, но я думаю, что он достигает того, что вы хотите сделать. Вы можете передать объект подключения, чтобы вы могли отправить ему флаг из MyHandler.

class MyHandler(PatternMatchingEventHandler):
    patterns = ["*.challenger", "*.challenger"]

    def __init__(self, conn, *args, **kwargs):
        super(MyHandler, self).__init__(*args, **kwargs)
        self.conn_list = conn

    def process(self, event):
        time.sleep(5)
        print event.src_path, event.event_type  # print now only for degug

    def on_created(self, event):
        self.process(event)
        c = self.conn_list.pop()
        c.send('1')
        c.close()



s = socket.socket()         # Create a socket object
host = socket.gethostname()  # Get local machine name
port = 12346                # Reserve a port for your service.
s.bind((host, port))        # Bind to the port
flag = '0'
s.listen(5)                 # Now wait for client connection.


from collections import deque
conn_list = deque()

if __name__ == '__main__':
    observer = Observer()
    observer.start()
    args = sys.argv[1:]
    observer.schedule(MyHandler(conn_list), path='/home/abcd/Maildir/new')
while True:
    try:
        c, addr = s.accept()     # Establish connection with client.
        print 'Got connection from', addr
        conn_list.append(c)

    except KeyboardInterrupt:
        observer.stop()
        while conn_list:
            conn_list.pop().close()
        print "Connections closed"
        break

observer.join()