<s> from casexml . apps . case . models import CommCareCase <EOL> from corehq . apps . change_feed import data_sources <EOL> from corehq . apps . change_feed . connection import get_kafka_client_or_none <EOL> from corehq . apps . change_feed . document_types import get_doc_meta_object_from_document , change_meta_from_doc_meta_and_document <EOL> from corehq . apps . change_feed . exceptions import MissingMetaInformationError <EOL> from corehq . apps . change_feed . producer import ChangeProducer <EOL> from corehq . apps . change_feed . topics import get_topic <EOL> from corehq . apps . domain . models import Domain <EOL> from corehq . apps . users . models import CommCareUser <EOL> from corehq . util . couchdb_management import couch_config <EOL> from pillowtop . checkpoints . manager import PillowCheckpoint , PillowCheckpointEventHandler <EOL> from pillowtop . feed . couch import CouchChangeFeed <EOL> from pillowtop . pillow . interface import ConstructedPillow <EOL> from pillowtop . processors import PillowProcessor <EOL> class KafkaProcessor ( PillowProcessor ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , kafka , data_source_type , data_source_name ) : <EOL> self . _kafka = kafka <EOL> self . _producer = ChangeProducer ( self . _kafka ) <EOL> self . _data_source_type = data_source_type <EOL> self . _data_source_name = data_source_name <EOL> def process_change ( self , pillow_instance , change , do_set_checkpoint = False ) : <EOL> try : <EOL> document = change . get_document ( ) <EOL> doc_meta = get_doc_meta_object_from_document ( document ) <EOL> change_meta = change_meta_from_doc_meta_and_document ( <EOL> doc_meta = doc_meta , <EOL> document = document , <EOL> data_source_type = self . _data_source_type , <EOL> data_source_name = self . _data_source_name , <EOL> doc_id = change . id , <EOL> ) <EOL> except MissingMetaInformationError : <EOL> pass <EOL> else : <EOL> change_meta . is_deletion = change_meta . is_deletion or change . deleted <EOL> self . _producer . send_change ( get_topic ( doc_meta ) , change_meta ) <EOL> def get_default_couch_db_change_feed_pillow ( pillow_id ) : <EOL> return get_change_feed_pillow_for_db ( pillow_id , CommCareCase . get_db ( ) ) <EOL> def get_user_groups_db_kafka_pillow ( pillow_id ) : <EOL> return get_change_feed_pillow_for_db ( pillow_id , couch_config . get_db_for_class ( CommCareUser ) ) <EOL> def get_domain_db_kafka_pillow ( pillow_id ) : <EOL> return get_change_feed_pillow_for_db ( pillow_id , couch_config . get_db_for_class ( Domain ) ) <EOL> def get_change_feed_pillow_for_db ( pillow_id , couch_db ) : <EOL> kafka_client = get_kafka_client_or_none ( ) <EOL> processor = KafkaProcessor ( <EOL> kafka_client , data_source_type = data_sources . COUCH , data_source_name = couch_db . dbname <EOL> ) <EOL> checkpoint = PillowCheckpoint ( pillow_id ) <EOL> return ConstructedPillow ( <EOL> name = pillow_id , <EOL> checkpoint = checkpoint , <EOL> change_feed = CouchChangeFeed ( couch_db , include_docs = True ) , <EOL> processor = processor , <EOL> change_processed_event_handler = PillowCheckpointEventHandler ( <EOL> checkpoint = checkpoint , checkpoint_frequency = <NUM_LIT:100> , <EOL> ) , <EOL> ) </s>
