def kafka_consumer_signalling_topic_handler(signalling_topic_data):
    logger.info('Creating Kafka Consumer for %s topic', signalling_topic_data['topic'])
    consumer = KafkaConsumer(signalling_topic_data['topic'], bootstrap_servers=[(dcm_ip_address + ':9092')], auto_offset_reset='earliest', enable_auto_commit=True, group_id=None, value_deserializer=(lambda x: json.loads(x.decode('utf-8'))))
    while signalling_start:
        message = consumer.poll(timeout_ms=1000)
        if (message != {}):
            logger.info('Message received in %s topic: %s', signalling_topic_data['topic'], message)
            for (tp, messages) in message.items():
                for msg in messages:
                    logger.info('Value: %s', msg.value)
                    topic = json.loads(msg.value)['topic']
                    if (json.loads(msg.value)['action'] == 'subscribe'):
                        logger.info('Create Logstash pipeline for topic %s', topic)
                        subprocess.call(['/bin/bash', '/usr/bin/dcs/create_logstash_pipeline.sh', topic])
                        thread = threading.Thread(target=kafka_consumer_refresh_dashboard_handler, args=[topic, msg.value])
                        thread.start()
                        try:
                            connection = psycopg2.connect(user='eve', password=eve_db_password, host='localhost', port='5432', dbname='pipelines')
                            logger.info('Inserting %s topic in database', topic)
                            cursor = connection.cursor()
                            cursor.execute('INSERT INTO pipeline VALUES ( %s )', (topic,))
                            connection.commit()
                            logger.info('Topic %s inserted in database', topic)
                            cursor.close()
                            connection.close()
                        except (Exception, psycopg2.Error) as error:
                            logger.error('Error while connecting to PostgreSQL: ', error)
                    elif (json.loads(msg.value)['action'] == 'unsubscribe'):
                        logger.info('Delete Logstash pipeline for topic %s', topic)
                        subprocess.call(['/bin/bash', '/usr/bin/dcs/delete_logstash_pipeline.sh', topic, 'no'])
                        scheduled_thread = threading.Timer(timedelta(days=14).total_seconds(), index_cleaner, args=[topic, msg.value])
                        scheduled_thread.start()
                        logger.info('Data removal for topic %s scheduled in 14 days', topic)
                        try:
                            connection = psycopg2.connect(user='eve', password=eve_db_password, host='localhost', port='5432', dbname='pipelines')
                            logger.info('Deleting %s topic in database', topic)
                            cursor = connection.cursor()
                            cursor.execute('DELETE FROM pipeline WHERE topic = %s', (topic,))
                            connection.commit()
                            logger.info('Topic %s deleted in database', topic)
                            cursor.close()
                            connection.close()
                        except (Exception, psycopg2.Error) as error:
                            logger.error('Error while connecting to PostgreSQL: ', error)
                    else:
                        logger.error('Action not allowed')
    logger.info('Closing Kafka Consumer for %s topic', signalling_topic_data['topic'])
    consumer.close()
def kafka_consumer_signalling_topic_handler(signalling_topic_data):
    logger.info('Creating Kafka Consumer for %s topic', signalling_topic_data['topic'])
    consumer = KafkaConsumer(signalling_topic_data['topic'], bootstrap_servers=[(dcm_ip_address + ':9092')], auto_offset_reset='earliest', enable_auto_commit=True, group_id=None, value_deserializer=(lambda x: json.loads(x.decode('utf-8'))))
    while signalling_start:
        message = consumer.poll(timeout_ms=1000)
        if (message != {}):
            logger.info('Message received in %s topic: %s', signalling_topic_data['topic'], message)
            for (tp, messages) in message.items():
                for msg in messages:
                    logger.info('Value: %s', msg.value)
                    topic = json.loads(msg.value)['topic']
                    if (json.loads(msg.value)['action'] == 'subscribe'):
                        logger.info('Create Logstash pipeline for topic %s', topic)
                        subprocess.call(['/bin/bash', '/usr/bin/dcs/create_logstash_pipeline.sh', topic])
                        thread = threading.Thread(target=kafka_consumer_refresh_dashboard_handler, args=[topic, msg.value])
                        thread.start()
                        try:
                            connection = psycopg2.connect(user='eve', password=eve_db_password, host='localhost', port='5432', dbname='pipelines')
                            logger.info('Inserting %s topic in database', topic)
                            cursor = connection.cursor()
                            cursor.execute('INSERT INTO pipeline VALUES ( %s )', (topic,))
                            connection.commit()
                            logger.info('Topic %s inserted in database', topic)
                            cursor.close()
                            connection.close()
                        except (Exception, psycopg2.Error) as error:
                            logger.error('Error while connecting to PostgreSQL: ', error)
                    elif (json.loads(msg.value)['action'] == 'unsubscribe'):
                        logger.info('Delete Logstash pipeline for topic %s', topic)
                        subprocess.call(['/bin/bash', '/usr/bin/dcs/delete_logstash_pipeline.sh', topic, 'no'])
                        scheduled_thread = threading.Timer(timedelta(days=14).total_seconds(), index_cleaner, args=[topic, msg.value])
                        scheduled_thread.start()
                        logger.info('Data removal for topic %s scheduled in 14 days', topic)
                        try:
                            connection = psycopg2.connect(user='eve', password=eve_db_password, host='localhost', port='5432', dbname='pipelines')
                            logger.info('Deleting %s topic in database', topic)
                            cursor = connection.cursor()
                            cursor.execute('DELETE FROM pipeline WHERE topic = %s', (topic,))
                            connection.commit()
                            logger.info('Topic %s deleted in database', topic)
                            cursor.close()
                            connection.close()
                        except (Exception, psycopg2.Error) as error:
                            logger.error('Error while connecting to PostgreSQL: ', error)
                    else:
                        logger.error('Action not allowed')
    logger.info('Closing Kafka Consumer for %s topic', signalling_topic_data['topic'])
    consumer.close()