fix #17

Merged
emmatveev merged 1 commits from queues into dev 2024-11-22 01:32:49 +03:00
5 changed files with 28 additions and 31 deletions

View File

@ -9,7 +9,8 @@ services:
environment: environment:
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
STAGE: "development" STAGE: "development"
QUEUES_TOKEN: $QUEUES_TOKEN_DEV networks:
- queues
deploy: deploy:
mode: replicated mode: replicated
restart_policy: restart_policy:
@ -25,8 +26,9 @@ services:
MONGO_HOST: "mongo.develop.sprinthub.ru" MONGO_HOST: "mongo.develop.sprinthub.ru"
MONGO_PASSWORD: $MONGO_PASSWORD_DEV MONGO_PASSWORD: $MONGO_PASSWORD_DEV
STAGE: "development" STAGE: "development"
QUEUES_TOKEN: $QUEUES_TOKEN_DEV
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
networks:
- queues
deploy: deploy:
mode: replicated mode: replicated
restart_policy: restart_policy:
@ -41,7 +43,8 @@ services:
environment: environment:
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV TELEGRAM_TOKEN: $TELEGRAM_TOKEN_DEV
STAGE: "development" STAGE: "development"
QUEUES_TOKEN: $QUEUES_TOKEN_DEV networks:
- queues
deploy: deploy:
mode: replicated mode: replicated
restart_policy: restart_policy:
@ -72,3 +75,5 @@ services:
networks: networks:
common-infra-nginx: common-infra-nginx:
external: true external: true
queues:
external: true

View File

@ -9,7 +9,8 @@ services:
environment: environment:
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
STAGE: "production" STAGE: "production"
QUEUES_TOKEN: $QUEUES_TOKEN_PROD networks:
- queues
deploy: deploy:
mode: replicated mode: replicated
restart_policy: restart_policy:
@ -25,8 +26,9 @@ services:
MONGO_HOST: "mongo.sprinthub.ru" MONGO_HOST: "mongo.sprinthub.ru"
MONGO_PASSWORD: $MONGO_PASSWORD_PROD MONGO_PASSWORD: $MONGO_PASSWORD_PROD
STAGE: "production" STAGE: "production"
QUEUES_TOKEN: $QUEUES_TOKEN_PROD
PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN PLATFORM_SECURITY_TOKEN: $PLATFORM_SECURITY_TOKEN
networks:
- queues
deploy: deploy:
mode: replicated mode: replicated
restart_policy: restart_policy:
@ -41,7 +43,8 @@ services:
environment: environment:
TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD TELEGRAM_TOKEN: $TELEGRAM_TOKEN_PROD
STAGE: "production" STAGE: "production"
QUEUES_TOKEN: $QUEUES_TOKEN_PROD networks:
- queues
deploy: deploy:
mode: replicated mode: replicated
restart_policy: restart_policy:
@ -72,3 +75,5 @@ services:
networks: networks:
common-infra-nginx: common-infra-nginx:
external: true external: true
queues:
external: true

View File

@ -42,5 +42,4 @@ jobs:
TELEGRAM_TOKEN_DEV: ${{ secrets.TELEGRAM_TOKEN_DEV }} TELEGRAM_TOKEN_DEV: ${{ secrets.TELEGRAM_TOKEN_DEV }}
MONGO_PASSWORD_DEV: ${{ secrets.MONGO_PASSWORD_DEV }} MONGO_PASSWORD_DEV: ${{ secrets.MONGO_PASSWORD_DEV }}
PLATFORM_SECURITY_TOKEN: ${{ secrets.PLATFORM_SECURITY_TOKEN }} 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 run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-dev.yaml pizda-bot

View File

@ -42,5 +42,4 @@ jobs:
TELEGRAM_TOKEN_PROD: ${{ secrets.TELEGRAM_TOKEN_PROD }} TELEGRAM_TOKEN_PROD: ${{ secrets.TELEGRAM_TOKEN_PROD }}
MONGO_PASSWORD_PROD: ${{ secrets.MONGO_PASSWORD_PROD }} MONGO_PASSWORD_PROD: ${{ secrets.MONGO_PASSWORD_PROD }}
PLATFORM_SECURITY_TOKEN: ${{ secrets.PLATFORM_SECURITY_TOKEN }} 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 run: docker stack deploy --with-registry-auth -c ./.deploy/deploy-prod.yaml pizda-bot

View File

@ -5,14 +5,10 @@ import time
stage = os.getenv("STAGE", 'local') stage = os.getenv("STAGE", 'local')
if stage == 'development': if stage == 'local':
QUEUES_URL = 'https://queues.develop.sprinthub.ru' QUEUES_URL = 'http://localhost:1239'
elif stage == 'production':
QUEUES_URL = 'https://queues.sprinthub.ru'
else: else:
QUEUES_URL = None QUEUES_URL = 'http://queues:1239'
token = os.getenv('QUEUES_TOKEN')
class QueuesException(Exception): class QueuesException(Exception):
@ -22,25 +18,18 @@ class QueuesException(Exception):
class TasksHandlerMixin: class TasksHandlerMixin:
def poll(self): def poll(self):
while True: while True:
if QUEUES_URL is None: response = requests.get(f'{QUEUES_URL}/api/v1/take', headers={'queue': self.queue_name})
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: if response.status_code == 404:
time.sleep(0.2) time.sleep(0.2)
continue continue
if response.status_code == 403:
raise NotImplemented('QUEUE_TOKEN is incorrect')
data = response.json() data = response.json()
try: try:
self.process(data['payload']) self.process(data['payload'])
except Exception as exc: except Exception as exc:
print(f'Error processing message id={data["id"]}, payload={data["payload"]}, exc={exc}') print(f'Error processing message id={data["id"]}, payload={data["payload"]}, exc={exc}')
continue continue
if QUEUES_URL is None:
continue
try: 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: if resp.status_code != 202:
raise QueuesException raise QueuesException
except: except:
@ -55,7 +44,7 @@ class TasksHandlerMixin:
def set_task(queue_name: str, payload: dict, seconds_to_execute: int, delay: int|None = None): 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, 'payload': payload,
'seconds_to_execute': seconds_to_execute, 'seconds_to_execute': seconds_to_execute,
'delay': delay, 'delay': delay,