<commit_before>#!/usr/bin/env python
# -*- coding: utf-8 -*-
from tornado.options import define,options
from stormed import Connection,Message
import settings
import logging
from tornado.ioloop import IOLoop
import json
import pycurl
import StringIO
import subprocess
from util.oauth import GitLabOAuth2Mixin
from services.oauth import OAuthService
from services.service import ServiceService
from services.application import ApplicationService
from services.application_access import ApplicationAccessService
from tornado.httputil import HTTPHeaders
from tornado import gen
import md5
import time
import random
import os
import tarfile
from tornado.httpclient import AsyncHTTPClient
from docker.utils import create_host_config
from docker.errors import APIError
conn = Connection(host=settings.MQ_HOST,username=settings.MQ_USERNAME,
      password=<PASSWORD>,heartbeat=settings.MQ_HEARTBEAT)
msg = Message('Hello World!')

def callback(msg):
    logging.info(" test=> Received %r" % msg.body)

def on_connect():
    ch = conn.channel()
    ch.queue_declare(queue='hello')
    ch.publish(msg, exchange='', routing_key='hello')
    logging.info(" test=> Sent 'Hello World!'")
    ch.consume('hello', callback, no_ack=True)
    
    logging.info("Init amqp success")
    options.mq_connection = conn
    # 初始化队列
    init_queue()
    # 初始化消费者
    init_consumer()

def init_queue():
    ch = conn.channel()
    
    ch.exchange_declare(exchange= settings.CREATE_SERVICE_EXCHANGE , type='direct',durable=True)
    ch.queue_declare(queue=settings.CREATE_SERVICE_QUEUE,durable=True)
    ch.queue_bind(queue=settings.CREATE_SERVICE_QUEUE,
                  exchange=settings.CREATE_SERVICE_EXCHANGE,
                  routing_key=settings.CREATE_SERVICE_ROUTING)

    ch.exchange_declare(exchange= settings.RUN_APPLICATION_EXCHANGE , type='direct',durable=True)
    ch.queue_declare(queue=settings.RUN_APPLICATION_QUEUE,durable=True)
    ch.queue_bind(queue=settings.RUN_APPLICATION_QUEUE,
                  exchange=settings.RUN_APPLICATION_EXCHANGE,
                  routing_key=settings.RUN_APPLICATION_ROUTING)

    ch.exchange_declare(exchange= settings.ACCESS_APPLICATION_EXCHANGE , type='topic',durable=True)
    ch.queue_declare(queue=settings.ACCESS_APPLICATION_QUEUE,durable=True)
    ch.queue_bind(queue=settings.ACCESS_APPLICATION_QUEUE,
                  exchange=settings.ACCESS_APPLICATION_EXCHANGE,
                  routing_key=settings.ACCESS_APPLICATION_ROUTING+"."+settings.CURRENT_HOST)

    logging.info("Declare amqp queue and exchange")
    
def init_consumer():
    ch = conn.channel()
    ch.consume(settings.CREATE_SERVICE_QUEUE,create_service,no_ack=False)
    ch.consume(settings.RUN_APPLICATION_QUEUE,run_application,no_ack=False)
    ch.consume(settings.ACCESS_APPLICATION_QUEUE,access_application,no_ack=False)
    logging.info("Init amqp consumer success")

def init_amqp():
    def handle_error(conn_error):
            logging.error("Init amqp error")
    conn.on_error = handle_error
    
    def handle_disconnect():
            logging.error("Init amqp failed")
            options.ioloop.stop()
    conn.on_disconnect = handle_disconnect
    conn.connect(on_connect)

def send_message(message,exchange_name,routing_key):
    ch = conn.channel()
    ch.publish(msg, exchange=exchange_name, routing_key=routing_key)

def create_service(msg):
    build_context = json.loads(msg.body)
    builder = BuildImage(build_context)
    builder.start_build_context()
    msg.ack()

def run_application(msg):
def run_application(msg):
    start_context = json.loads(msg.body)
    builder = StartContainer(start_context)
    builder.start_run_application()
    msg.ack()

def access_application(msg):
    access_context = json.loads(msg.body)
    builder = AccessContainer(access_context)
    builder.start_access_application()
    msg.ack()

class BuildImage():
    s_service = ServiceService()
    def __init__(self,build_context):<commit_msg>[BUGFIX]consumer bug<commit_after>#!/usr/bin/env python
# -*- coding: utf-8 -*-
from tornado.options import define,options
from stormed import Connection,Message
import settings
import logging
from tornado.ioloop import IOLoop
import json
import pycurl
import StringIO
import subprocess
from util.oauth import GitLabOAuth2Mixin
from services.oauth import OAuthService
from services.service import ServiceService
from services.application import ApplicationService
from services.application_access import ApplicationAccessService
from tornado.httputil import HTTPHeaders
from tornado import gen
import md5
import time
import random
import os
import tarfile
from tornado.httpclient import AsyncHTTPClient
from docker.utils import create_host_config
from docker.errors import APIError
conn = Connection(host=settings.MQ_HOST,username=settings.MQ_USERNAME,
      password=settings.MQ_PASSWORD,heartbeat=settings.MQ_HEARTBEAT)
msg = Message('Hello World!')

def callback(msg):
    logging.info(" test=> Received %r" % msg.body)

def on_connect():
    ch = conn.channel()
    ch.queue_declare(queue='hello')
    ch.publish(msg, exchange='', routing_key='hello')
    logging.info(" test=> Sent 'Hello World!'")
    ch.consume('hello', callback, no_ack=True)
    
    logging.info("Init amqp success")
    options.mq_connection = conn
    # 初始化队列
    init_queue()
    # 初始化消费者
    init_consumer()

def init_queue():
    ch = conn.channel()
    
    ch.exchange_declare(exchange= settings.CREATE_SERVICE_EXCHANGE , type='direct',durable=True)
    ch.queue_declare(queue=settings.CREATE_SERVICE_QUEUE,durable=True)
    ch.queue_bind(queue=settings.CREATE_SERVICE_QUEUE,
                  exchange=settings.CREATE_SERVICE_EXCHANGE,
                  routing_key=settings.CREATE_SERVICE_ROUTING)

    ch.exchange_declare(exchange= settings.RUN_APPLICATION_EXCHANGE , type='direct',durable=True)
    ch.queue_declare(queue=settings.RUN_APPLICATION_QUEUE,durable=True)
    ch.queue_bind(queue=settings.RUN_APPLICATION_QUEUE,
                  exchange=settings.RUN_APPLICATION_EXCHANGE,
                  routing_key=settings.RUN_APPLICATION_ROUTING)

    ch.exchange_declare(exchange= settings.ACCESS_APPLICATION_EXCHANGE , type='topic',durable=True)
    ch.queue_declare(queue=settings.ACCESS_APPLICATION_QUEUE,durable=True)
    ch.queue_bind(queue=settings.ACCESS_APPLICATION_QUEUE,
                  exchange=settings.ACCESS_APPLICATION_EXCHANGE,
                  routing_key=settings.ACCESS_APPLICATION_ROUTING+"."+settings.CURRENT_HOST)

    logging.info("Declare amqp queue and exchange")
    
def init_consumer():
    ch = conn.channel()
    ch.consume(settings.CREATE_SERVICE_QUEUE,create_service,no_ack=False)
    ch.consume(settings.RUN_APPLICATION_QUEUE,run_application,no_ack=False)
    ch.consume(settings.ACCESS_APPLICATION_QUEUE,access_application,no_ack=False)
    logging.info("Init amqp consumer success")

def init_amqp():
    def handle_error(conn_error):
            logging.error("Init amqp error")
    conn.on_error = handle_error
    
    def handle_disconnect():
            logging.error("Init amqp failed")
            options.ioloop.stop()
    conn.on_disconnect = handle_disconnect
    conn.connect(on_connect)

def send_message(message,exchange_name,routing_key):
    ch = conn.channel()
    ch.publish(msg, exchange=exchange_name, routing_key=routing_key)

def create_service(msg):
    build_context = json.loads(msg.body)
    builder = BuildImage(build_context)
    builder.start_build_context()
    msg.ack()

def run_application(msg):
    start_context = json.loads(msg.body)
    builder = StartContainer(start_context)
    builder.start_run_application()
    msg.ack()

def access_application(msg):
    access_context = json.loads(msg.body)
    builder = AccessContainer(access_context)
    builder.start_access_application()
    msg.ack()

class BuildImage():
    s_service = ServiceService()
    def __init__(self,build_context):
{'fraction_non_alphanumeric': 0.09245926487305797, 'fraction_numerical': 0.0007578628268283441, 'mean_word_length': 4.61958836053939, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1728144', 'n_tokens_mistral': 2465, 'n_tokens_neox': 2390, 'n_words': 426}