Merge pull request 'fix' (#18) from queues into master
Reviewed-on: #18
This commit is contained in:
commit
37ac151e68
@ -9,7 +9,8 @@ services:
|
||||
environment:
|
||||
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
|
||||
STAGE: "development"
|
||||
QUEUES_TOKEN: $QUEUES_TOKEN_DEV
|
||||
networks:
|
||||
- queues
|
||||
deploy:
|
||||
mode: replicated
|
||||
restart_policy:
|
||||
@ -25,8 +26,9 @@ services:
|
||||
MONGO_HOST: "mongo.develop.sprinthub.ru"
|
||||
MONGO_PASSWORD: $MONGO_PASSWORD_DEV
|
||||
STAGE: "development"
|
||||
QUEUES_TOKEN: $QUEUES_TOKEN_DEV
|
||||
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
|
||||
networks:
|
||||
- queues
|
||||
deploy:
|
||||
mode: replicated
|
||||
restart_policy:
|
||||
@ -41,7 +43,8 @@ services:
|
||||
environment:
|
||||
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
|
||||
STAGE: "development"
|
||||
QUEUES_TOKEN: $QUEUES_TOKEN_DEV
|
||||
networks:
|
||||
- queues
|
||||
deploy:
|
||||
mode: replicated
|
||||
restart_policy:
|
||||
@ -71,4 +74,6 @@ services:
|
||||
|
||||
networks:
|
||||
common-infra-nginx:
|
||||
external: true
|
||||
external: true
|
||||
queues:
|
||||
external: true
|
||||
|
@ -9,7 +9,8 @@ services:
|
||||
environment:
|
||||
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
|
||||
STAGE: "production"
|
||||
QUEUES_TOKEN: $QUEUES_TOKEN_PROD
|
||||
networks:
|
||||
- queues
|
||||
deploy:
|
||||
mode: replicated
|
||||
restart_policy:
|
||||
@ -25,8 +26,9 @@ services:
|
||||
MONGO_HOST: "mongo.sprinthub.ru"
|
||||
MONGO_PASSWORD: $MONGO_PASSWORD_PROD
|
||||
STAGE: "production"
|
||||
QUEUES_TOKEN: $QUEUES_TOKEN_PROD
|
||||
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
|
||||
networks:
|
||||
- queues
|
||||
deploy:
|
||||
mode: replicated
|
||||
restart_policy:
|
||||
@ -41,7 +43,8 @@ services:
|
||||
environment:
|
||||
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
|
||||
STAGE: "production"
|
||||
QUEUES_TOKEN: $QUEUES_TOKEN_PROD
|
||||
networks:
|
||||
- queues
|
||||
deploy:
|
||||
mode: replicated
|
||||
restart_policy:
|
||||
@ -71,4 +74,6 @@ services:
|
||||
|
||||
networks:
|
||||
common-infra-nginx:
|
||||
external: true
|
||||
external: true
|
||||
queues:
|
||||
external: true
|
||||
|
@ -42,5 +42,4 @@ jobs:
|
||||
TELEGRAM_TOKEN_DEV: ${{ secrets.TELEGRAM_TOKEN_DEV }}
|
||||
MONGO_PASSWORD_DEV: ${{ secrets.MONGO_PASSWORD_DEV }}
|
||||
PLATFORM_SECURITY_TOKEN: ${{ secrets.PLATFORM_SECURITY_TOKEN }}
|
||||
QUEUES_TOKEN_DEV: ${{ secrets.QUEUES_TOKEN_DEV }}
|
||||
run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-dev.yaml pizda-bot
|
||||
|
@ -42,5 +42,4 @@ jobs:
|
||||
TELEGRAM_TOKEN_PROD: ${{ secrets.TELEGRAM_TOKEN_PROD }}
|
||||
MONGO_PASSWORD_PROD: ${{ secrets.MONGO_PASSWORD_PROD }}
|
||||
PLATFORM_SECURITY_TOKEN: ${{ secrets.PLATFORM_SECURITY_TOKEN }}
|
||||
QUEUES_TOKEN_PROD: ${{ secrets.QUEUES_TOKEN_PROD }}
|
||||
run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-prod.yaml pizda-bot
|
||||
|
@ -5,14 +5,10 @@ import time
|
||||
|
||||
|
||||
stage = os.getenv("STAGE", 'local')
|
||||
if stage == 'development':
|
||||
QUEUES_URL = 'https://queues.develop.sprinthub.ru'
|
||||
elif stage == 'production':
|
||||
QUEUES_URL = 'https://queues.sprinthub.ru'
|
||||
if stage == 'local':
|
||||
QUEUES_URL = 'http://localhost:1239'
|
||||
else:
|
||||
QUEUES_URL = None
|
||||
|
||||
token = os.getenv('QUEUES_TOKEN')
|
||||
QUEUES_URL = 'http://queues:1239'
|
||||
|
||||
|
||||
class QueuesException(Exception):
|
||||
@ -22,25 +18,18 @@ class QueuesException(Exception):
|
||||
class TasksHandlerMixin:
|
||||
def poll(self):
|
||||
while True:
|
||||
if QUEUES_URL is None:
|
||||
data = {'payload': json.loads(input('Input message: '))}
|
||||
else:
|
||||
response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name, 'X-Queues-Token': token})
|
||||
if response.status_code == 404:
|
||||
time.sleep(0.2)
|
||||
continue
|
||||
if response.status_code == 403:
|
||||
raise NotImplemented('QUEUE_TOKEN is incorrect')
|
||||
data = response.json()
|
||||
response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name})
|
||||
if response.status_code == 404:
|
||||
time.sleep(0.2)
|
||||
continue
|
||||
data = response.json()
|
||||
try:
|
||||
self.process(data['payload'])
|
||||
except Exception as exc:
|
||||
print(f'Error processing message id={data["id"]}, payload={data["payload"]}, exc={exc}')
|
||||
continue
|
||||
if QUEUES_URL is None:
|
||||
continue
|
||||
try:
|
||||
resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': data['id']}, headers={'X-Queues-Token': token})
|
||||
resp = requests.post(f'{QUEUES_URL}/api/v1/finish', json={'id': data['id']})
|
||||
if resp.status_code != 202:
|
||||
raise QueuesException
|
||||
except:
|
||||
@ -55,7 +44,7 @@ class TasksHandlerMixin:
|
||||
|
||||
|
||||
def set_task(queue_name: str, payload: dict, seconds_to_execute: int, delay: int|None = None):
|
||||
resp = requests.post(f'{QUEUES_URL}/api/v1/put', headers={'queue': queue_name, 'X-Queues-Token': token}, json={
|
||||
resp = requests.post(f'{QUEUES_URL}/api/v1/put', headers={'queue': queue_name}, json={
|
||||
'payload': payload,
|
||||
'seconds_to_execute': seconds_to_execute,
|
||||
'delay': delay,
|
||||
|
Loading…
Reference in New Issue
Block a user