001/* 002 * (C) Copyright 2006-2016 Nuxeo SA (http://nuxeo.com/) and others. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 * 016 * 017 * Contributors: 018 * anechaev 019 */ 020package org.nuxeo.ecm.platform.importer.mqueues.kafka; 021 022import org.apache.commons.logging.Log; 023import org.apache.commons.logging.LogFactory; 024import org.nuxeo.ecm.core.event.EventServiceComponent; 025import org.nuxeo.runtime.model.ComponentContext; 026import org.nuxeo.runtime.model.ComponentInstance; 027import org.nuxeo.runtime.model.DefaultComponent; 028 029import java.util.HashMap; 030import java.util.Map; 031import java.util.Properties; 032import java.util.Set; 033 034public class KafkaConfigServiceImpl extends DefaultComponent implements KafkaConfigService { 035 private static final Log log = LogFactory.getLog(KafkaConfigServiceImpl.class); 036 public static final String KAFKA_CONFIG_XP = "kafkaConfig"; 037 protected Map<String, KafkaConfigDescriptor> configs = new HashMap<>(); 038 039 @Override 040 public void registerContribution(Object contribution, String extensionPoint, ComponentInstance contributor) { 041 if (extensionPoint.equals(KAFKA_CONFIG_XP)) { 042 KafkaConfigDescriptor descriptor = (KafkaConfigDescriptor) contribution; 043 configs.put(descriptor.name, descriptor); 044 log.info(String.format("Register Kafka contribution: %s", descriptor.name)); 045 } 046 } 047 048 @Override 049 public int getApplicationStartedOrder() { 050 // since there is no deps, let's start before WorkManager 051 return EventServiceComponent.APPLICATION_STARTED_ORDER - 10; 052 } 053 054 @Override 055 public void deactivate(ComponentContext context) { 056 super.deactivate(context); 057 log.debug("Deactivating service"); 058 } 059 060 @Override 061 public void activate(ComponentContext context) { 062 super.activate(context); 063 log.debug("Activating service"); 064 } 065 066 public Set<String> getConfigNames() { 067 return configs.keySet(); 068 } 069 070 @Override 071 public String getZkServers(String configName) { 072 checkConfigName(configName); 073 return configs.get(configName).zkServers; 074 } 075 076 protected void checkConfigName(String configName) { 077 if (!configs.containsKey(configName)) { 078 throw new IllegalArgumentException("Unknown configuration name: " + configName); 079 } 080 } 081 082 @Override 083 public Properties getProducerProperties(String configName) { 084 checkConfigName(configName); 085 return configs.get(configName).getProducerProperties(); 086 } 087 088 @Override 089 public Properties getConsumerProperties(String configName) { 090 checkConfigName(configName); 091 return configs.get(configName).getConsumerProperties(); 092 } 093 094 @Override 095 public String getTopicPrefix(String configName) { 096 checkConfigName(configName); 097 return configs.get(configName).getTopicPrefix(); 098 } 099}