fetch emails through Zero MQ
This commit is contained in:
		
							parent
							
								
									8c5d4d7301
								
							
						
					
					
						commit
						47bda266a4
					
				
					 10 changed files with 62 additions and 27 deletions
				
			
		
							
								
								
									
										6
									
								
								app/conf/config.py
									
										
									
									
									
										Normal file
									
								
							
							
						
						
									
										6
									
								
								app/conf/config.py
									
										
									
									
									
										Normal file
									
								
							|  | @ -0,0 +1,6 @@ | |||
| #!/usr/bin/env python | ||||
| # -*- coding: utf-8 -*- | ||||
| 
 | ||||
| # TODO move to JSON config | ||||
| 
 | ||||
| zmq = {'pub_port': 7701, 'sub_port':7702} | ||||
							
								
								
									
										43
									
								
								app/interface/zclient.py
									
										
									
									
									
										Normal file
									
								
							
							
						
						
									
										43
									
								
								app/interface/zclient.py
									
										
									
									
									
										Normal file
									
								
							|  | @ -0,0 +1,43 @@ | |||
| #!/usr/bin/env python | ||||
| # -*- coding: utf-8 -*- | ||||
| 
 | ||||
| import zmq | ||||
| from conf import config | ||||
| from threading import Thread | ||||
| import logging | ||||
| import json | ||||
| from app.services import processor | ||||
| 
 | ||||
| logger = logging.getLogger(__name__) | ||||
| 
 | ||||
| context = zmq.Context() | ||||
| 
 | ||||
| 
 | ||||
| def process(message): | ||||
|     data = json.loads(message) | ||||
|     if data['topic'] == 'email:newmail': | ||||
|         logger.info('newmail => {}'.format(data)) | ||||
|         processor.enqueue({'request': 'new_mail', 'data': data}) | ||||
| 
 | ||||
| 
 | ||||
| class Consumer(Thread): | ||||
| 
 | ||||
|     def run(self): | ||||
|         zsub = context.socket(zmq.SUB) | ||||
|         zsub.connect('tcp://127.0.0.1:{}'.format(config.zmq['pub_port'])) | ||||
|         zsub.setsockopt_string(zmq.SUBSCRIBE, '') | ||||
|         self.loop = True | ||||
|         while self.loop: | ||||
|             message = zsub.recv() | ||||
|             try: | ||||
|                 process(message) | ||||
|             except: | ||||
|                 logger.exception('cannot process broker message') | ||||
| 
 | ||||
|     def stop(self): | ||||
|         self.loop = False | ||||
| 
 | ||||
| 
 | ||||
| def start(): | ||||
|     c = Consumer() | ||||
|     c.start() | ||||
							
								
								
									
										25
									
								
								app/run.py
									
										
									
									
									
								
							
							
						
						
									
										25
									
								
								app/run.py
									
										
									
									
									
								
							|  | @ -18,10 +18,11 @@ for path in paths: | |||
| import config | ||||
| from app.services import database | ||||
| from app.services import processor | ||||
| from app.controllers import api | ||||
| from app.controllers import form | ||||
| from app.controllers import report | ||||
| from app.controllers import mail | ||||
| from app.interface import api | ||||
| from app.interface import form | ||||
| from app.interface import report | ||||
| #from app.controllers import mail | ||||
| from app.interface import zclient | ||||
| from app import app | ||||
| 
 | ||||
| 
 | ||||
|  | @ -45,20 +46,10 @@ configure_logging(logging_level) | |||
| logger = logging.getLogger(__name__) | ||||
| 
 | ||||
| # initialize database | ||||
| database.setup() | ||||
| database.setup()   | ||||
| 
 | ||||
| #from app.helpers.hashing import md5 | ||||
| #from app.models.comment import Comment | ||||
| #for comment in Comment.select(): | ||||
| #    email = comment.author_email.strip().lower() | ||||
| #    if email: | ||||
| #        comment.author_gravatar = md5(email) | ||||
| #    comment.author_email = '' | ||||
| #    comment.save()    | ||||
| 
 | ||||
| # routes | ||||
| logger.debug('imported: %s ' % api.__name__) | ||||
| logger.debug('imported: %s ' % mail.__name__) | ||||
| # start broker client | ||||
| zclient.start() | ||||
| 
 | ||||
| # start processor | ||||
| template_path = os.path.abspath(os.path.join(current_path, 'templates')) | ||||
|  |  | |||
|  | @ -14,7 +14,6 @@ from app.models.reader import Reader | |||
| from app.models.report import Report | ||||
| from app.models.comment import Comment | ||||
| from app.helpers.hashing import md5 | ||||
| import requests | ||||
| import json | ||||
| import config | ||||
| import PyRSS2Gen | ||||
|  | @ -392,7 +391,8 @@ def mail(to_email, subject, message): | |||
|         'subject': subject, | ||||
|         'content': message | ||||
|     } | ||||
|     r = requests.post(config.MAIL_URL, data=json.dumps(msg), headers=headers) | ||||
|     # do something smart here | ||||
|     # r = requests.post(config.MAIL_URL, data=json.dumps(msg), headers=headers) | ||||
|     if r.status_code in (200, 201): | ||||
|         logger.debug('Email for %s posted' % to_email) | ||||
|     else: | ||||
|  |  | |||
|  | @ -1,16 +1,11 @@ | |||
| certifi==2017.11.5 | ||||
| chardet==3.0.4 | ||||
| click==6.7 | ||||
| Flask==0.12.2 | ||||
| Flask-Cors==3.0.3 | ||||
| idna==2.6 | ||||
| itsdangerous==0.24 | ||||
| Jinja2==2.10 | ||||
| Markdown==2.6.9 | ||||
| Markdown==2.6.11 | ||||
| MarkupSafe==1.0 | ||||
| peewee==2.10.2 | ||||
| PyRSS2Gen==1.1 | ||||
| requests==2.18.4 | ||||
| six==1.11.0 | ||||
| urllib3==1.22 | ||||
| Werkzeug==0.12.2 | ||||
| pyzmq==16.0.3 | ||||
| Werkzeug==0.14.1 | ||||
|  |  | |||
		Loading…
	
	Add table
		
		Reference in a new issue
	
	 Yax
						Yax