UNOMI-101 : New feature to add the ability to import profiles
Project: http://git-wip-us.apache.org/repos/asf/incubator-unomi/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-unomi/commit/89d4c8eb Tree: http://git-wip-us.apache.org/repos/asf/incubator-unomi/tree/89d4c8eb Diff: http://git-wip-us.apache.org/repos/asf/incubator-unomi/diff/89d4c8eb Branch: refs/heads/master Commit: 89d4c8eb9646705cfd6c4bdb79fdac27ee9a5634 Parents: 0f44140 Author: Abdelkader Midani <amid...@apache.org> Authored: Tue May 23 02:30:28 2017 +0200 Committer: Abdelkader Midani <amid...@apache.org> Committed: Mon Jun 12 20:09:59 2017 +0200 ---------------------------------------------------------------------- extensions/pom.xml | 1 + extensions/router/README.md | 113 ++++++++++ extensions/router/pom.xml | 64 ++++++ extensions/router/router-api/pom.xml | 43 ++++ .../unomi/router/api/ImportConfiguration.java | 223 +++++++++++++++++++ .../unomi/router/api/ProfileToImport.java | 77 +++++++ .../services/ImportConfigurationService.java | 60 +++++ .../api/services/ProfileImportService.java | 29 +++ extensions/router/router-core/pom.xml | 182 +++++++++++++++ .../core/context/ProfileImportCamelContext.java | 170 ++++++++++++++ .../core/processor/ConfigUpdateProcessor.java | 44 ++++ .../ImportConfigByFileNameProcessor.java | 44 ++++ .../core/processor/LineSplitProcessor.java | 114 ++++++++++ .../core/processor/UnomiStorageProcessor.java | 46 ++++ .../ProfileImportConfigUpdateRouteBuilder.java | 62 ++++++ .../ProfileImportKafkaToUnomiRouteBuilder.java | 77 +++++++ .../route/ProfileImportOneShotRouteBuilder.java | 99 ++++++++ .../ProfileImportSourceToKafkaRouteBuilder.java | 120 ++++++++++ .../resources/OSGI-INF/blueprint/blueprint.xml | 94 ++++++++ .../main/resources/org.apache.unomi.router.cfg | 25 +++ extensions/router/router-karaf-feature/pom.xml | 157 +++++++++++++ extensions/router/router-rest/pom.xml | 75 +++++++ .../ImportConfigurationServiceEndPoint.java | 176 +++++++++++++++ .../resources/OSGI-INF/blueprint/blueprint.xml | 80 +++++++ extensions/router/router-service/pom.xml | 104 +++++++++ .../ImportConfigurationServiceImpl.java | 114 ++++++++++ .../services/ProfileImportServiceImpl.java | 121 ++++++++++ .../resources/OSGI-INF/blueprint/blueprint.xml | 39 ++++ itests/pom.xml | 8 +- package/pom.xml | 11 + performance-tests/pom.xml | 4 +- ...g.apache.unomi.persistence.elasticsearch.cfg | 2 +- persistence-elasticsearch/plugins/pom.xml | 2 +- samples/trainingplugin.zip | Bin 0 -> 48508 bytes samples/trainingplugin/pom.xml | 55 +++++ .../training/TrainedNotificationAction.java | 61 +++++ .../cxs/actions/trainingNotifAction.json | 13 ++ .../META-INF/cxs/rules/trainedNotification.json | 20 ++ .../resources/OSGI-INF/blueprint/blueprint.xml | 34 +++ 39 files changed, 2755 insertions(+), 8 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/pom.xml ---------------------------------------------------------------------- diff --git a/extensions/pom.xml b/extensions/pom.xml index 7126b61..dea4bc2 100644 --- a/extensions/pom.xml +++ b/extensions/pom.xml @@ -34,6 +34,7 @@ <module>lists-extension</module> <module>privacy-extension</module> <module>geonames</module> + <module>router</module> </modules> </project> http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/README.md ---------------------------------------------------------------------- diff --git a/extensions/router/README.md b/extensions/router/README.md new file mode 100644 index 0000000..0d382d4 --- /dev/null +++ b/extensions/router/README.md @@ -0,0 +1,113 @@ +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one or more + ~ contributor license agreements. See the NOTICE file distributed with + ~ this work for additional information regarding copyright ownership. + ~ The ASF licenses this file to You under the Apache License, Version 2.0 + ~ (the "License"); you may not use this file except in compliance with + ~ the License. You may obtain a copy of the License at + ~ + ~ http://www.apache.org/licenses/LICENSE-2.0 + ~ + ~ Unless required by applicable law or agreed to in writing, software + ~ distributed under the License is distributed on an "AS IS" BASIS, + ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + ~ See the License for the specific language governing permissions and + ~ limitations under the License. + --> + +Unomi Router +========================== + +## Getting started +Unomi Router Extension a Karaf Feature that provide an Enterprise Application Integration tool. +It is optional so you must configure it and install it in Karaf, and can be used for Machine - Machine or Human - Machine integration with Unomi. +Mainly Unomi Router Extension aim to make it easy to import third party applications/platforms profiles into Unomi. +This extension is implemented using Apache Camel routes and is using Apache Kafka to buffer import process and make it failsafe. + +## Getting started +1. Configure your Unomi Router: + In the `etc/org.apache.unomi.router.cfg` file, you might want to update the following settings: + Kafka settings + >`#Kafka settings` + + >`kafka.host=localhost` + + >`kafka.port=9092` + + >`kafka.import.topic=camel-deposit` + + >`kafka.import.groupId=unomi-import-group` + + Kafka host and port with the topic name and the groupId ti which the topic is assigned + + >`#Import One Shot upload directory` + + >`import.oneshot.uploadDir=/tmp/unomi_oneshot_import_configs/` + + Path to the folder where unomi should stock file imported for a oneshot processing + + +2. Deploy into Apache Unomi using the following commands from the Apache Karaf shell: + ```sh + $ feature:repo-add mvn:org.apache.unomi/unomi-router-karaf-feature/${version}/xml/features + $ feature:install unomi-router-karaf-feature + ``` + +3. Send your import configuration: + + An import configuration is nothing else than a simple JSON to describe how you want to import your data (Profiles). + To create/update an import configuration + + `POST /cxs/importConfiguration` + ```json + { + "itemId": "f57f1f86-97bf-4ba0-b4e4-7d5e77e7c0bd", + "itemType": "importConfig", + "scope": "integration", + "name": "Test Recurrent", + "description": "Just test recurrent import", + "configType": "recurrent", + "properties": { + "source": "{file/ftp}://{path}?fileName={file-name}.csv&move=.done&consumer.delay=20000", + "mapping": { + "firstName": 0, + "lastName": 1, + ... + } + }, + "mergingProperty": "email", + "overwriteExistingProfiles": true, + "propertiesToOverwrite": ["firstName", "lastName"], + "active": true + } + ``` + + Omit the `itemId` when creating new entry, `configType` can be '**recurrent**' for file/ftp/network path polling or '**oneshot**' for one time import. + + The `properties.source` attribute is an Apache Camel endpoint uri (See http://camel.apache.org/uris.html for more details). Unomi Router is designed to use **File** and **FTP** Camel components. + + The attribute `properties.mapping` is a Map of: + * Key: Profile property id in Unomi + * Value: Index of the column in the imported file to copy the in the previous property. + + The attribute `mergingProperty` is the profile property id in Unomi to use to check for duplication. + + The attribute `propertiesToOverwrite` is a list of profile properties ids to overwrite, if **null** all properties + will be overwritten. + + The attribute `active` is the flag to activate or deactivate the import configuration. + + Concerning oneshot import configuration using the previously described service will only create the import configuration, to send the file to process + you need to call : + + `POST /cxs/importConfiguration/oneshot` + + `Content-Type : multipart/form-data` + + First multipart with the name '**importConfigId**' is the importConfiguration to use to import the file, second one with the name '**file**' is the file to import. + + + + + + \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/pom.xml ---------------------------------------------------------------------- diff --git a/extensions/router/pom.xml b/extensions/router/pom.xml new file mode 100644 index 0000000..c69320c --- /dev/null +++ b/extensions/router/pom.xml @@ -0,0 +1,64 @@ +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one or more + ~ contributor license agreements. See the NOTICE file distributed with + ~ this work for additional information regarding copyright ownership. + ~ The ASF licenses this file to You under the Apache License, Version 2.0 + ~ (the "License"); you may not use this file except in compliance with + ~ the License. You may obtain a copy of the License at + ~ + ~ http://www.apache.org/licenses/LICENSE-2.0 + ~ + ~ Unless required by applicable law or agreed to in writing, software + ~ distributed under the License is distributed on an "AS IS" BASIS, + ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + ~ See the License for the specific language governing permissions and + ~ limitations under the License. + --> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.unomi</groupId> + <artifactId>unomi-extensions</artifactId> + <version>1.2.0-incubating-dmf_1343-SNAPSHOT</version> + </parent> + + <artifactId>unomi-router</artifactId> + <name>Apache Unomi :: Extensions :: Router</name> + <description>Apache Camel Router for the Apache Unomi Context server</description> + <packaging>pom</packaging> + + <properties> + <camel.version>2.18.3</camel.version> + </properties> + + <build> + <plugins> + <plugin> + <groupId>org.apache.felix</groupId> + <artifactId>maven-bundle-plugin</artifactId> + <extensions>true</extensions> + <configuration> + <instructions> + <Embed-Dependency>*;scope=compile|runtime</Embed-Dependency> + <Import-Package> + sun.misc;resolution:=optional, + * + </Import-Package> + </instructions> + </configuration> + </plugin> + </plugins> + </build> + + <modules> + <module>router-api</module> + <module>router-service</module> + <module>router-core</module> + <module>router-rest</module> + <module>router-karaf-feature</module> + </modules> + +</project> http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-api/pom.xml ---------------------------------------------------------------------- diff --git a/extensions/router/router-api/pom.xml b/extensions/router/router-api/pom.xml new file mode 100644 index 0000000..06207b3 --- /dev/null +++ b/extensions/router/router-api/pom.xml @@ -0,0 +1,43 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one or more + ~ contributor license agreements. See the NOTICE file distributed with + ~ this work for additional information regarding copyright ownership. + ~ The ASF licenses this file to You under the Apache License, Version 2.0 + ~ (the "License"); you may not use this file except in compliance with + ~ the License. You may obtain a copy of the License at + ~ + ~ http://www.apache.org/licenses/LICENSE-2.0 + ~ + ~ Unless required by applicable law or agreed to in writing, software + ~ distributed under the License is distributed on an "AS IS" BASIS, + ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + ~ See the License for the specific language governing permissions and + ~ limitations under the License. + --> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>unomi-router</artifactId> + <groupId>org.apache.unomi</groupId> + <version>1.2.0-incubating-dmf_1343-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>unomi-router-api</artifactId> + <name>Apache Unomi :: Extensions :: Router :: API</name> + <description>Router Specification API</description> + <packaging>bundle</packaging> + + <dependencies> + <dependency> + <groupId>org.apache.unomi</groupId> + <artifactId>unomi-api</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> + </dependencies> + + +</project> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/ImportConfiguration.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/ImportConfiguration.java b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/ImportConfiguration.java new file mode 100644 index 0000000..127de39 --- /dev/null +++ b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/ImportConfiguration.java @@ -0,0 +1,223 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.api; + +import org.apache.unomi.api.Item; +import org.apache.unomi.api.MetadataItem; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Created by amidani on 28/04/2017. + */ +public class ImportConfiguration extends Item { + + /** + * The ImportConfiguration ITEM_TYPE + * + * @see Item for a discussion of ITEM_TYPE + */ + public static final String ITEM_TYPE = "importConfig"; + private String name; + private String description; + private String configType; + private Map<String, Object> properties = new HashMap<>(); + private String mergingProperty; + private boolean overwriteExistingProfiles = false; + private List<String> propertiesToOverwrite; + + private String columnSeparator = ","; + private String lineSeparator = "\n"; + private boolean active = false; + + /** + * Sets the property identified by the specified name to the specified value. If a property with that name already exists, replaces its value, otherwise adds the new + * property with the specified name and value. + * + * @param name the name of the property to set + * @param value the value of the property + */ + public void setProperty(String name, Object value) { + properties.put(name, value); + } + + /** + * Retrieves the name of the import configuration + * @return the name of the import configuration + */ + public String getName() { return this.name; } + + /** + * Sets the name of the import configuration + * @param name the name of the import configuration + */ + public void setName(String name) { + this.name = name; + } + + /** + * Retrieves the description of the import configuration + * @return the description of the import configuration + */ + public String getDescription() { return this.description; } + + /** + * Sets the description of the import configuration + * @param description the description of the import configuration + */ + public void setDescription(String description) { + this.description = description; + } + + + /** + * Retrieves the config type of the import configuration + * @return the config type of the import configuration + */ + public String getConfigType() { return this.configType; } + + /** + * Sets the config type of the import configuration + * @param configType the config type of the import configuration + */ + public void setConfigType(String configType) { + this.configType = configType; + } + + /** + * Retrieves the property identified by the specified name. + * + * @param name the name of the property to retrieve + * @return the value of the specified property or {@code null} if no such property exists + */ + public Object getProperty(String name) { + return properties.get(name); + } + + /** + * Retrieves a Map of all property name - value pairs for this import configuration. + * + * @return a Map of all property name - value pairs for this import configuration + */ + public Map<String, Object> getProperties() { + return properties; + } + + /** + * Sets the property name - value pairs for this import configuration. + * + * @param properties a Map containing the property name - value pairs for this import configuration + */ + public void setProperties(Map<String, Object> properties) { + this.properties = properties; + } + + public String getMergingProperty() { + return mergingProperty; + } + + /** + * Sets the merging property. + * @param mergingProperty property used to check if the profile exist when merging + */ + public void setMergingProperty(String mergingProperty) { + this.mergingProperty = mergingProperty; + } + + + /** + * Retrieves the import configuration active flag. + * + * @return true if the import configuration is active false if not + */ + public boolean isActive() { + return this.active; + } + + /** + * Sets the active flag true/false. + * + * @param active a boolean to set to active or inactive the import configuration + */ + public void setActive(boolean active) { + this.active = active; + } + + /** + * Retrieves the import configuration overwriteExistingProfiles flag. + * + * @return true if during the import existing profiles must be overwritten + */ + public boolean isOverwriteExistingProfiles() { + return this.overwriteExistingProfiles; + } + + /** + * Sets the overwriteExistingProfiles flag true/false. + * + * @param overwriteExistingProfiles a boolean to set overwriteExistingProfiles in the import configuration + */ + public void setOverwriteExistingProfiles(boolean overwriteExistingProfiles) { + this.overwriteExistingProfiles = overwriteExistingProfiles; + } + + public List<String> getPropertiesToOverwrite() { + return propertiesToOverwrite; + } + + public void setPropertiesToOverwrite(List<String> propertiesToOverwrite) { + this.propertiesToOverwrite = propertiesToOverwrite; + } + + /** + * gets the column separator. + */ + public String getColumnSeparator() { + return this.columnSeparator; + } + + /** + * Sets the column separator. + * @param columnSeparator property used to specify a line separator. Defaults to ',' + */ + public void setColumnSeparator(String columnSeparator) { + if(this.columnSeparator !=null) { + this.columnSeparator = columnSeparator; + } + } + + /** + * gets the line separator. + */ + public String getLineSeparator() { + return this.lineSeparator; + } + + /** + * Sets the line separator. + * @param lineSeparator property used to specify a line separator. Defaults to '\n' + */ + public void setLineSeparator(String lineSeparator) { + if(lineSeparator != null) { + this.lineSeparator = lineSeparator; + } + } + + +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/ProfileToImport.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/ProfileToImport.java b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/ProfileToImport.java new file mode 100644 index 0000000..30e40e0 --- /dev/null +++ b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/ProfileToImport.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.unomi.router.api; + +import org.apache.unomi.api.Profile; + +import java.util.List; + +/** + * An extension of {@link Profile} to handle merge strategy and deletion when importing profiles + */ +public class ProfileToImport extends Profile { + + private List<String> propertiesToOverwrite; + private String mergingProperty; + private boolean profileToDelete; + private boolean overwriteExistingProfiles; + + + public List<String> getPropertiesToOverwrite() { + return this.propertiesToOverwrite; + } + + public void setPropertiesToOverwrite(List<String> propertiesToOverwrite) { + this.propertiesToOverwrite = propertiesToOverwrite; + } + + public boolean isProfileToDelete() { + return this.profileToDelete; + } + + public void setProfileToDelete(boolean profileToDelete) { + this.profileToDelete = profileToDelete; + } + + public boolean isOverwriteExistingProfiles() { + return this.overwriteExistingProfiles; + } + + /** + * Sets the overwriteExistingProfiles flag. + * @param overwriteExistingProfiles flag used to specify if we want to overwrite existing profiles + */ + public void setOverwriteExistingProfiles(boolean overwriteExistingProfiles) { + this.overwriteExistingProfiles = overwriteExistingProfiles; + } + + public String getMergingProperty() { + return this.mergingProperty; + } + + /** + * Sets the merging property. + * @param mergingProperty property used to check if the profile exist when merging + */ + public void setMergingProperty(String mergingProperty) { + this.mergingProperty = mergingProperty; + } + + + +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ImportConfigurationService.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ImportConfigurationService.java b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ImportConfigurationService.java new file mode 100644 index 0000000..cacd671 --- /dev/null +++ b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ImportConfigurationService.java @@ -0,0 +1,60 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.api.services; + +import org.apache.unomi.router.api.ImportConfiguration; + +import java.util.List; + +/** + * A service to access and operate on {@link ImportConfiguration}s. + */ +public interface ImportConfigurationService { + + /** + * Retrieves all the import configurations. + * + * @return the list of import configurations + */ + public List<ImportConfiguration> getImportConfigurations(); + + /** + * Retrieves the import configuration identified by the specified identifier. + * + * @param configId the identifier of the profile to retrieve + * @return the import configuration identified by the specified identifier or + * {@code null} if no such import configuration exists + */ + public ImportConfiguration load(String configId); + + /** + * Saves the specified import configuration in the context server. + * + * @param profile the import configuration to be saved + * @return the newly saved import configuration + */ + public ImportConfiguration save(ImportConfiguration profile); + + /** + * Deletes the import configuration identified by the specified identifier. + * + * @param configId the identifier of the import configuration to delete + */ + public void delete(String configId); + + +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ProfileImportService.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ProfileImportService.java b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ProfileImportService.java new file mode 100644 index 0000000..aa7d182 --- /dev/null +++ b/extensions/router/router-api/src/main/java/org/apache/unomi/router/api/services/ProfileImportService.java @@ -0,0 +1,29 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.api.services; + +import org.apache.unomi.router.api.ProfileToImport; + +import java.lang.reflect.InvocationTargetException; + +/** + * Created by amidani on 20/05/2017. + */ +public interface ProfileImportService { + + boolean saveMergeDeleteImportedProfile(ProfileToImport profileToImport) throws InvocationTargetException, IllegalAccessException; +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/pom.xml ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/pom.xml b/extensions/router/router-core/pom.xml new file mode 100644 index 0000000..53780e7 --- /dev/null +++ b/extensions/router/router-core/pom.xml @@ -0,0 +1,182 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one or more + ~ contributor license agreements. See the NOTICE file distributed with + ~ this work for additional information regarding copyright ownership. + ~ The ASF licenses this file to You under the Apache License, Version 2.0 + ~ (the "License"); you may not use this file except in compliance with + ~ the License. You may obtain a copy of the License at + ~ + ~ http://www.apache.org/licenses/LICENSE-2.0 + ~ + ~ Unless required by applicable law or agreed to in writing, software + ~ distributed under the License is distributed on an "AS IS" BASIS, + ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + ~ See the License for the specific language governing permissions and + ~ limitations under the License. + --> +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <parent> + <artifactId>unomi-router</artifactId> + <groupId>org.apache.unomi</groupId> + <version>1.2.0-incubating-dmf_1343-SNAPSHOT</version> + </parent> + <modelVersion>4.0.0</modelVersion> + + <artifactId>unomi-router-core</artifactId> + <name>Apache Unomi :: Extensions :: Router :: Core</name> + <description>Router Core (Apache Camel Routes)</description> + <packaging>bundle</packaging> + + <dependencies> + <dependency> + <groupId>org.osgi</groupId> + <artifactId>org.osgi.core</artifactId> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.osgi</groupId> + <artifactId>org.osgi.compendium</artifactId> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.unomi</groupId> + <artifactId>unomi-api</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.unomi</groupId> + <artifactId>unomi-services</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.unomi</groupId> + <artifactId>unomi-router-api</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>org.apache.camel</groupId> + <artifactId>camel-core</artifactId> + <version>${camel.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.camel</groupId> + <artifactId>camel-jackson</artifactId> + <version>${camel.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.camel</groupId> + <artifactId>camel-http-common</artifactId> + <version>${camel.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.camel</groupId> + <artifactId>camel-servlet</artifactId> + <version>${camel.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.camel</groupId> + <artifactId>camel-kafka</artifactId> + <version>${camel.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.commons</groupId> + <artifactId>commons-lang3</artifactId> + </dependency> + <dependency> + <groupId>commons-net</groupId> + <artifactId>commons-net</artifactId> + <version>3.5</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.servicemix.bundles</groupId> + <artifactId>org.apache.servicemix.bundles.jsch</artifactId> + <version>0.1.54_1</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>org.apache.kafka</groupId> + <artifactId>kafka-clients</artifactId> + <version>0.10.1.0</version> + <scope>provided</scope> + </dependency> + </dependencies> + + <build> + <plugins> + <plugin> + <groupId>org.apache.felix</groupId> + <artifactId>maven-bundle-plugin</artifactId> + <extensions>true</extensions> + <configuration> + <instructions> + <Embed-Dependency>*;scope=compile|runtime</Embed-Dependency> + <Import-Package> + org.osgi.service.event;resolution:=optional, + org.apache.camel, + org.apache.camel.builder, + org.apache.camel.component.file.remote, + org.apache.camel.component.file, + org.apache.camel.component.jackson, + org.apache.camel.component.kafka, + org.apache.camel.component.servlet, + org.apache.camel.component.servlet.osgi, + org.apache.camel.impl, + org.apache.camel.model, + org.apache.camel.model.dataformat, + org.apache.camel.model.rest, + org.apache.camel.spi, + org.apache.unomi.api, + org.apache.unomi.router.api, + org.apache.unomi.api.services, + org.apache.unomi.router.api.services, + org.apache.kafka.clients.producer;resolution:=optional, + org.apache.kafka.clients.consumer;resolution:=optional, + com.jcraft.jsch, + org.osgi.framework, + org.osgi.service.http, + org.slf4j + </Import-Package> + </instructions> + </configuration> + </plugin> + <plugin> + <groupId>org.codehaus.mojo</groupId> + <artifactId>build-helper-maven-plugin</artifactId> + <executions> + <execution> + <id>attach-artifacts</id> + <phase>package</phase> + <goals> + <goal>attach-artifact</goal> + </goals> + <configuration> + <artifacts> + <artifact> + <file> + src/main/resources/org.apache.unomi.router.cfg + </file> + <type>cfg</type> + <classifier>routercfg</classifier> + </artifact> + </artifacts> + </configuration> + </execution> + </executions> + </plugin> + </plugins> + </build> + + + +</project> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/ProfileImportCamelContext.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/ProfileImportCamelContext.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/ProfileImportCamelContext.java new file mode 100644 index 0000000..df734d3 --- /dev/null +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/context/ProfileImportCamelContext.java @@ -0,0 +1,170 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.core.context; + +import org.apache.camel.CamelContext; +import org.apache.camel.Route; +import org.apache.camel.component.jackson.JacksonDataFormat; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.unomi.router.api.ImportConfiguration; +import org.apache.unomi.router.api.services.ImportConfigurationService; +import org.apache.unomi.router.core.processor.ImportConfigByFileNameProcessor; +import org.apache.unomi.router.core.processor.UnomiStorageProcessor; +import org.apache.unomi.router.core.route.ProfileImportKafkaToUnomiRouteBuilder; +import org.apache.unomi.router.core.route.ProfileImportOneShotRouteBuilder; +import org.apache.unomi.router.core.route.ProfileImportSourceToKafkaRouteBuilder; +import org.osgi.framework.Bundle; +import org.osgi.framework.BundleContext; +import org.osgi.framework.BundleEvent; +import org.osgi.framework.SynchronousBundleListener; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * Created by amidani on 04/05/2017. + */ +public class ProfileImportCamelContext implements SynchronousBundleListener { + + private Logger logger = LoggerFactory.getLogger(ProfileImportCamelContext.class.getName()); + + private CamelContext camelContext; + private UnomiStorageProcessor unomiStorageProcessor; + private ImportConfigByFileNameProcessor importConfigByFileNameProcessor; + private ImportConfigurationService importConfigurationService; + private JacksonDataFormat jacksonDataFormat; + private String uploadDir; + private Map<String, String> kafkaProps; + + private final String IMPORT_CONFIG_TYPE_RECURRENT = "recurrent"; + + private BundleContext bundleContext; + + public void setBundleContext(BundleContext bundleContext) { + this.bundleContext = bundleContext; + } + + public void initCamelContext() throws Exception { + logger.info("Initialize Camel Context..."); + camelContext = new DefaultCamelContext(); + List<ImportConfiguration> importConfigurationList = importConfigurationService.getImportConfigurations(); + ProfileImportSourceToKafkaRouteBuilder builderReader = new ProfileImportSourceToKafkaRouteBuilder(kafkaProps); + builderReader.setImportConfigurationList(importConfigurationList); + builderReader.setJacksonDataFormat(jacksonDataFormat); + builderReader.setContext(camelContext); + camelContext.addRoutes(builderReader); + + //One shot import route + ProfileImportOneShotRouteBuilder builderOneShot = new ProfileImportOneShotRouteBuilder(kafkaProps); + builderOneShot.setImportConfigByFileNameProcessor(importConfigByFileNameProcessor); + builderOneShot.setJacksonDataFormat(jacksonDataFormat); + builderOneShot.setUploadDir(uploadDir); + builderOneShot.setContext(camelContext); + camelContext.addRoutes(builderOneShot); + + + ProfileImportKafkaToUnomiRouteBuilder builderProcessor = new ProfileImportKafkaToUnomiRouteBuilder(kafkaProps); + builderProcessor.setUnomiStorageProcessor(unomiStorageProcessor); + builderProcessor.setJacksonDataFormat(jacksonDataFormat); + builderProcessor.setContext(camelContext); + camelContext.addRoutes(builderProcessor); + + camelContext.start(); + + logger.debug("postConstruct {" + bundleContext.getBundle() + "}"); + + processBundleStartup(bundleContext); + for (Bundle bundle : bundleContext.getBundles()) { + if (bundle.getBundleContext() != null) { + processBundleStartup(bundle.getBundleContext()); + } + } + bundleContext.addBundleListener(this); + logger.info("Camel Context {} initialized successfully."); + + } + + private boolean stopRoute(String routeId) throws Exception { + return camelContext.stopRoute(routeId, 10L, TimeUnit.SECONDS, true); + } + + public void updateProfileImportReaderRoute(ImportConfiguration importConfiguration) throws Exception { + Route route = camelContext.getRoute(importConfiguration.getItemId()); + if(route!=null && stopRoute(importConfiguration.getItemId())) { + camelContext.removeRoute(importConfiguration.getItemId()); + } + //Handle transforming an import config oneshot <--> recurrent + if(IMPORT_CONFIG_TYPE_RECURRENT.equals(importConfiguration.getConfigType())){ + ProfileImportSourceToKafkaRouteBuilder builder = new ProfileImportSourceToKafkaRouteBuilder(kafkaProps); + builder.setImportConfigurationList(Arrays.asList(importConfiguration)); + builder.setJacksonDataFormat(jacksonDataFormat); + builder.setContext(camelContext); + camelContext.addRoutes(builder); + } + } + + public CamelContext getCamelContext() { + return camelContext; + } + + public void setUnomiStorageProcessor(UnomiStorageProcessor unomiStorageProcessor) { + this.unomiStorageProcessor = unomiStorageProcessor; + } + + public void setImportConfigByFileNameProcessor(ImportConfigByFileNameProcessor importConfigByFileNameProcessor) { + this.importConfigByFileNameProcessor = importConfigByFileNameProcessor; + } + + public void setImportConfigurationService(ImportConfigurationService importConfigurationService) { + this.importConfigurationService = importConfigurationService; + } + + public void setJacksonDataFormat(JacksonDataFormat jacksonDataFormat) { + this.jacksonDataFormat = jacksonDataFormat; + } + + public void setUploadDir(String uploadDir) { + this.uploadDir = uploadDir; + } + + public void setKafkaProps(Map<String, String> kafkaProps) { + this.kafkaProps = kafkaProps; + } + + public void preDestroy() throws Exception { + bundleContext.removeBundleListener(this); + //This is to shutdown Camel context + //(will stop all routes/components/endpoints etc and clear internal state/cache) + this.camelContext.stop(); + logger.info("Camel context for profile import is shutdown."); + } + + private void processBundleStartup(BundleContext bundleContext) { + if (bundleContext == null) { + return; + } + } + + @Override + public void bundleChanged(BundleEvent bundleEvent) { + + } +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/ConfigUpdateProcessor.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/ConfigUpdateProcessor.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/ConfigUpdateProcessor.java new file mode 100644 index 0000000..e4eaa19 --- /dev/null +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/ConfigUpdateProcessor.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.core.processor; + +import org.apache.camel.Exchange; +import org.apache.camel.Message; +import org.apache.camel.Processor; +import org.apache.unomi.router.api.ImportConfiguration; +import org.apache.unomi.router.core.context.ProfileImportCamelContext; + +/** + * Created by amidani on 10/05/2017. + */ +public class ConfigUpdateProcessor implements Processor{ + + private ProfileImportCamelContext profileImportCamelContext; + + @Override + public void process(Exchange exchange) throws Exception { + if (exchange.getIn() != null) { + Message message = exchange.getIn(); + ImportConfiguration importConfiguration = message.getBody(ImportConfiguration.class); + profileImportCamelContext.updateProfileImportReaderRoute(importConfiguration); + } + } + + public void setProfileImportCamelContext(ProfileImportCamelContext profileImportCamelContext) { + this.profileImportCamelContext = profileImportCamelContext; + } +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/ImportConfigByFileNameProcessor.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/ImportConfigByFileNameProcessor.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/ImportConfigByFileNameProcessor.java new file mode 100644 index 0000000..7fc7730 --- /dev/null +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/ImportConfigByFileNameProcessor.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.core.processor; + +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.component.file.GenericFile; +import org.apache.unomi.router.api.ImportConfiguration; +import org.apache.unomi.router.api.services.ImportConfigurationService; + +/** + * Created by amidani on 22/05/2017. + */ +public class ImportConfigByFileNameProcessor implements Processor{ + + private ImportConfigurationService importConfigurationService; + + @Override + public void process(Exchange exchange) throws Exception { + + String fileName = exchange.getIn().getBody(GenericFile.class).getFileName(); + String importConfigId = fileName.substring(0, fileName.indexOf('.')); + ImportConfiguration importConfiguration = importConfigurationService.load(importConfigId); + exchange.getIn().setHeader("importConfigOneShot", importConfiguration); + } + + public void setImportConfigurationService(ImportConfigurationService importConfigurationService) { + this.importConfigurationService = importConfigurationService; + } +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/LineSplitProcessor.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/LineSplitProcessor.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/LineSplitProcessor.java new file mode 100644 index 0000000..150ef6d --- /dev/null +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/LineSplitProcessor.java @@ -0,0 +1,114 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.core.processor; + +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.component.kafka.KafkaConstants; +import org.apache.commons.lang3.StringUtils; +import org.apache.unomi.router.api.ImportConfiguration; +import org.apache.unomi.router.api.ProfileToImport; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +/** + * Created by amidani on 29/12/2016. + */ +public class LineSplitProcessor implements Processor { + + private Map<String, Integer> fieldsMapping; + private List<String> propertiesToOverwrite; + private String mergingProperty; + private boolean overwriteExistingProfiles; + private String columnSeparator; + + @Override + public void process(Exchange exchange) throws Exception { + //In case of one shot import we check the header and overwrite import config + ImportConfiguration importConfigOneShot = (ImportConfiguration) exchange.getIn().getHeader("importConfigOneShot"); + if(importConfigOneShot!=null) { + fieldsMapping = (Map<String, Integer>)importConfigOneShot.getProperties().get("mapping"); + propertiesToOverwrite = importConfigOneShot.getPropertiesToOverwrite(); + mergingProperty = importConfigOneShot.getMergingProperty(); + overwriteExistingProfiles = importConfigOneShot.isOverwriteExistingProfiles(); + columnSeparator = importConfigOneShot.getColumnSeparator(); + } + String[] profileData = ((String)exchange.getIn().getBody()).split(columnSeparator); + ProfileToImport profileToImport = new ProfileToImport(); + profileToImport.setItemId(UUID.randomUUID().toString()); + profileToImport.setItemType("profile"); + profileToImport.setScope("system"); + if(profileData.length > 0) { + Map<String, Object> properties = new HashMap<>(); + for(String fieldMappingKey : fieldsMapping.keySet()) { + if(profileData.length > fieldsMapping.get(fieldMappingKey)) { + properties.put(fieldMappingKey, profileData[fieldsMapping.get(fieldMappingKey)].trim()); + } + } + profileToImport.setProperties(properties); + profileToImport.setMergingProperty(mergingProperty); + profileToImport.setPropertiesToOverwrite(propertiesToOverwrite); + profileToImport.setOverwriteExistingProfiles(overwriteExistingProfiles); + if(StringUtils.isNotBlank(profileData[profileData.length - 1]) && Boolean.parseBoolean(profileData[profileData.length - 1].trim())) { + profileToImport.setProfileToDelete(true); + } + } + exchange.getIn().setBody(profileToImport, ProfileToImport.class); + exchange.getIn().setHeader(KafkaConstants.PARTITION_KEY, 0); + exchange.getIn().setHeader(KafkaConstants.KEY, "1"); + } + + /** + * Setter of fieldsMapping + * @param fieldsMapping map String,Integer fieldName in ES and the matching column index in the import file + */ + public void setFieldsMapping(Map<String, Integer> fieldsMapping) { + this.fieldsMapping = fieldsMapping; + } + + public void setPropertiesToOverwrite(List<String> propertiesToOverwrite) { + this.propertiesToOverwrite = propertiesToOverwrite; + } + + public void setOverwriteExistingProfiles(boolean overwriteExistingProfiles) { + this.overwriteExistingProfiles = overwriteExistingProfiles; + } + + public String getMergingProperty() { + return this.mergingProperty; + } + + /** + * Sets the merging property. + * @param mergingProperty property used to check if the profile exist when merging + */ + public void setMergingProperty(String mergingProperty) { + this.mergingProperty = mergingProperty; + } + + /** + * Sets the line separator. + * @param columnSeparator property used to specify a line separator. Defaults to ',' + */ + public void setColumnSeparator(String columnSeparator) { + this.columnSeparator = columnSeparator; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/UnomiStorageProcessor.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/UnomiStorageProcessor.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/UnomiStorageProcessor.java new file mode 100644 index 0000000..7e55185 --- /dev/null +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/processor/UnomiStorageProcessor.java @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.core.processor; + +import org.apache.camel.Exchange; +import org.apache.camel.Message; +import org.apache.camel.Processor; +import org.apache.unomi.router.api.ProfileToImport; +import org.apache.unomi.router.api.services.ProfileImportService; + +/** + * Created by amidani on 29/12/2016. + */ +public class UnomiStorageProcessor implements Processor { + + private ProfileImportService profileImportService; + + @Override + public void process(Exchange exchange) + throws Exception { + if (exchange.getIn() != null) { + Message message = exchange.getIn(); + + ProfileToImport profileToImport = (ProfileToImport) message.getBody(); + profileImportService.saveMergeDeleteImportedProfile(profileToImport); + } + } + + public void setProfileImportService(ProfileImportService profileImportService) { + this.profileImportService = profileImportService; + } +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportConfigUpdateRouteBuilder.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportConfigUpdateRouteBuilder.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportConfigUpdateRouteBuilder.java new file mode 100644 index 0000000..40575d5 --- /dev/null +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportConfigUpdateRouteBuilder.java @@ -0,0 +1,62 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.core.route; + +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.model.rest.RestBindingMode; +import org.apache.unomi.router.api.ImportConfiguration; +import org.apache.unomi.router.core.context.ProfileImportCamelContext; +import org.apache.unomi.router.core.processor.ConfigUpdateProcessor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Created by amidani on 10/05/2017. + */ +public class ProfileImportConfigUpdateRouteBuilder extends RouteBuilder { + + private static final Logger logger = LoggerFactory.getLogger(ProfileImportConfigUpdateRouteBuilder.class.getName()); + + private ProfileImportCamelContext profileImportCamelContext; + + @Override + public void configure() throws Exception { + logger.info("Preparing REST Configuration for servlet with context path [/importConfigAdmin]"); + restConfiguration().component("servlet") + .contextPath("/importConfigAdmin") + .enableCORS(false) + .bindingMode(RestBindingMode.json) + .dataFormatProperty("prettyPrint", "true"); + + rest().put("/").consumes("application/json").type(ImportConfiguration.class) + .to("direct:importConfigRestDeposit"); + ConfigUpdateProcessor profileImportConfigUpdateProcessor = new ConfigUpdateProcessor(); + profileImportConfigUpdateProcessor.setProfileImportCamelContext(profileImportCamelContext); + from("direct:importConfigRestDeposit") + .process(profileImportConfigUpdateProcessor) + .transform().constant("Success.") + .onException(Exception.class) + .transform().constant("Failure!"); + + + } + + public void setProfileImportCamelContext(ProfileImportCamelContext profileImportCamelContext) { + this.profileImportCamelContext = profileImportCamelContext; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportKafkaToUnomiRouteBuilder.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportKafkaToUnomiRouteBuilder.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportKafkaToUnomiRouteBuilder.java new file mode 100644 index 0000000..1b056fe --- /dev/null +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportKafkaToUnomiRouteBuilder.java @@ -0,0 +1,77 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.core.route; + +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.jackson.JacksonDataFormat; +import org.apache.camel.component.kafka.KafkaComponent; +import org.apache.camel.component.kafka.KafkaConfiguration; +import org.apache.camel.component.kafka.KafkaEndpoint; +import org.apache.commons.lang3.StringUtils; +import org.apache.unomi.router.core.processor.UnomiStorageProcessor; + +import java.util.Map; + +/** + * Created by amidani on 26/04/2017. + */ +public class ProfileImportKafkaToUnomiRouteBuilder extends RouteBuilder { + + private UnomiStorageProcessor unomiStorageProcessor; + private JacksonDataFormat jacksonDataFormat; + private String kafkaHost; + private String kafkaPort; + private String kafkaImportTopic; + private String kafkaImportGroupId; + + public ProfileImportKafkaToUnomiRouteBuilder(Map<String, String> kafkaProps) { + kafkaHost = kafkaProps.get("kafkaHost"); + kafkaPort = kafkaProps.get("kafkaPort"); + kafkaImportTopic = kafkaProps.get("kafkaImportTopic"); + kafkaImportGroupId = kafkaProps.get("kafkaImportGroupId"); + } + + @Override + public void configure() throws Exception { + + StringBuilder kafkaUri = new StringBuilder("kafka:"); + kafkaUri.append(kafkaHost).append(":").append(kafkaPort).append("?topic=").append(kafkaImportTopic); + if(StringUtils.isNotBlank(kafkaImportGroupId)) { + kafkaUri.append("&groupId="+kafkaImportGroupId); + } + kafkaUri.append("&autoCommitEnable=true&consumersCount=10"); + KafkaConfiguration kafkaConfiguration = new KafkaConfiguration(); + kafkaConfiguration.setBrokers(kafkaHost+":"+kafkaPort); + kafkaConfiguration.setTopic(kafkaImportTopic); + kafkaConfiguration.setGroupId(kafkaImportGroupId); + KafkaEndpoint endpoint = new KafkaEndpoint(kafkaUri.toString(), new KafkaComponent(this.getContext())); + endpoint.setConfiguration(kafkaConfiguration); + from(endpoint) + .unmarshal(jacksonDataFormat) + .process(unomiStorageProcessor) + .to("log:org.apache.unomi.router?level=INFO"); + + } + + public void setUnomiStorageProcessor(UnomiStorageProcessor unomiStorageProcessor) { + this.unomiStorageProcessor = unomiStorageProcessor; + } + + public void setJacksonDataFormat(JacksonDataFormat jacksonDataFormat) { + this.jacksonDataFormat = jacksonDataFormat; + } +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportOneShotRouteBuilder.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportOneShotRouteBuilder.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportOneShotRouteBuilder.java new file mode 100644 index 0000000..d095f3e --- /dev/null +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportOneShotRouteBuilder.java @@ -0,0 +1,99 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.core.route; + +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.jackson.JacksonDataFormat; +import org.apache.camel.component.kafka.KafkaComponent; +import org.apache.camel.component.kafka.KafkaConfiguration; +import org.apache.camel.component.kafka.KafkaEndpoint; +import org.apache.commons.lang3.StringUtils; +import org.apache.unomi.router.core.processor.ImportConfigByFileNameProcessor; +import org.apache.unomi.router.core.processor.LineSplitProcessor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; + +/** + * Created by amidani on 22/05/2017. + */ +public class ProfileImportOneShotRouteBuilder extends RouteBuilder { + + private Logger logger = LoggerFactory.getLogger(ProfileImportOneShotRouteBuilder.class.getName()); + + private ImportConfigByFileNameProcessor importConfigByFileNameProcessor; + private JacksonDataFormat jacksonDataFormat; + private String uploadDir; + private String kafkaHost; + private String kafkaPort; + private String kafkaImportTopic; + private String kafkaImportGroupId; + + private final String IMPORT_ONESHOT_ROUTE_ID = "ONE_SHOT_ROUTE"; + + public ProfileImportOneShotRouteBuilder(Map<String, String> kafkaProps) { + kafkaHost = kafkaProps.get("kafkaHost"); + kafkaPort = kafkaProps.get("kafkaPort"); + kafkaImportTopic = kafkaProps.get("kafkaImportTopic"); + kafkaImportGroupId = kafkaProps.get("kafkaImportGroupId"); + } + + @Override + public void configure() throws Exception { + + //Prepare Kafka Deposit + StringBuilder kafkaUri = new StringBuilder("kafka:"); + kafkaUri.append(kafkaHost).append(":").append(kafkaPort).append("?topic=").append(kafkaImportTopic); + if(StringUtils.isNotBlank(kafkaImportGroupId)) { + kafkaUri.append("&groupId="+ kafkaImportGroupId); + } + + KafkaConfiguration kafkaConfiguration = new KafkaConfiguration(); + kafkaConfiguration.setBrokers(kafkaHost+":"+kafkaPort); + kafkaConfiguration.setTopic(kafkaImportTopic); + kafkaConfiguration.setGroupId(kafkaImportGroupId); + KafkaEndpoint endpoint = new KafkaEndpoint(kafkaUri.toString(), new KafkaComponent(this.getContext())); + endpoint.setConfiguration(kafkaConfiguration); + + LineSplitProcessor lineSplitProcessor = new LineSplitProcessor(); + + + from("file://"+uploadDir+"?include=.*.csv&consumer.delay=1m") + .routeId(IMPORT_ONESHOT_ROUTE_ID) + .autoStartup(true) + .process(importConfigByFileNameProcessor) + .split(bodyAs(String.class).tokenize("${in.header.importConfigOneShot.getLineSeparator}")) + .process(lineSplitProcessor) + .to("log:org.apache.unomi.router?level=INFO") + .marshal(jacksonDataFormat) + .convertBodyTo(String.class) + .to(endpoint); + } + + public void setImportConfigByFileNameProcessor(ImportConfigByFileNameProcessor importConfigByFileNameProcessor) { + this.importConfigByFileNameProcessor = importConfigByFileNameProcessor; + } + + public void setUploadDir(String uploadDir) { + this.uploadDir = uploadDir; + } + + public void setJacksonDataFormat(JacksonDataFormat jacksonDataFormat) { + this.jacksonDataFormat = jacksonDataFormat; + } +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportSourceToKafkaRouteBuilder.java ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportSourceToKafkaRouteBuilder.java b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportSourceToKafkaRouteBuilder.java new file mode 100644 index 0000000..37ae59e --- /dev/null +++ b/extensions/router/router-core/src/main/java/org/apache/unomi/router/core/route/ProfileImportSourceToKafkaRouteBuilder.java @@ -0,0 +1,120 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.unomi.router.core.route; + +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.jackson.JacksonDataFormat; +import org.apache.camel.component.kafka.KafkaComponent; +import org.apache.camel.component.kafka.KafkaConfiguration; +import org.apache.camel.component.kafka.KafkaEndpoint; +import org.apache.commons.lang3.StringUtils; +import org.apache.unomi.router.api.ImportConfiguration; +import org.apache.unomi.router.core.processor.LineSplitProcessor; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.Map; + +/** + * Created by amidani on 26/04/2017. + */ + +public class ProfileImportSourceToKafkaRouteBuilder extends RouteBuilder { + + private static final Logger logger = LoggerFactory.getLogger(ProfileImportSourceToKafkaRouteBuilder.class.getName()); + + private List<ImportConfiguration> importConfigurationList; + private JacksonDataFormat jacksonDataFormat; + private String kafkaHost; + private String kafkaPort; + private String kafkaImportTopic; + private String kafkaImportGroupId; + + public ProfileImportSourceToKafkaRouteBuilder(Map<String, String> kafkaProps) { + kafkaHost = kafkaProps.get("kafkaHost"); + kafkaPort = kafkaProps.get("kafkaPort"); + kafkaImportTopic = kafkaProps.get("kafkaImportTopic"); + kafkaImportGroupId = kafkaProps.get("kafkaImportGroupId"); + } + + @Override + public void configure() throws Exception { + //Prepare Kafka Deposit + StringBuilder kafkaUri = new StringBuilder("kafka:"); + kafkaUri.append(kafkaHost).append(":").append(kafkaPort).append("?topic=").append(kafkaImportTopic); + if(StringUtils.isNotBlank(kafkaImportGroupId)) { + kafkaUri.append("&groupId="+ kafkaImportGroupId); + } + + KafkaConfiguration kafkaConfiguration = new KafkaConfiguration(); + kafkaConfiguration.setBrokers(kafkaHost+":"+kafkaPort); + kafkaConfiguration.setTopic(kafkaImportTopic); + kafkaConfiguration.setGroupId(kafkaImportGroupId); + KafkaEndpoint endpoint = new KafkaEndpoint(kafkaUri.toString(), new KafkaComponent(this.getContext())); + endpoint.setConfiguration(kafkaConfiguration); + + //Loop on multiple import configuration + for(ImportConfiguration importConfiguration : importConfigurationList) { + if(importConfiguration.getProperties().size() > 0 && + StringUtils.isNotEmpty((String) importConfiguration.getProperties().get("source"))) { + //Prepare Split Processor + LineSplitProcessor lineSplitProcessor = new LineSplitProcessor(); + lineSplitProcessor.setFieldsMapping((Map<String, Integer>) importConfiguration.getProperties().get("mapping")); + lineSplitProcessor.setOverwriteExistingProfiles(importConfiguration.isOverwriteExistingProfiles()); + lineSplitProcessor.setPropertiesToOverwrite(importConfiguration.getPropertiesToOverwrite()); + lineSplitProcessor.setMergingProperty(importConfiguration.getMergingProperty()); + lineSplitProcessor.setColumnSeparator(importConfiguration.getColumnSeparator()); + + from((String) importConfiguration.getProperties().get("source")) + .routeId(importConfiguration.getItemId())// This allow identification of the route for manual start/stop + .autoStartup(importConfiguration.isActive())// Auto-start if the import configuration is set active + .split(bodyAs(String.class).tokenize(importConfiguration.getLineSeparator())) + .process(lineSplitProcessor) + .to("log:org.apache.unomi.router?level=INFO") + .marshal(jacksonDataFormat) + .convertBodyTo(String.class) + .to(endpoint); + } + } + } + + public void setImportConfigurationList(List<ImportConfiguration> importConfigurationList) { + this.importConfigurationList = importConfigurationList; + } + + public void setJacksonDataFormat(JacksonDataFormat jacksonDataFormat) { + this.jacksonDataFormat = jacksonDataFormat; + } + + public void setKafkaHost(String kafkaHost) { + this.kafkaHost = kafkaHost; + } + + public void setKafkaPort(String kafkaPort) { + this.kafkaPort = kafkaPort; + } + + public void setKafkaImportTopic(String kafkaImportTopic) { + this.kafkaImportTopic = kafkaImportTopic; + } + + public void setKafkaImportGroupId(String kafkaImportGroupId) { + this.kafkaImportGroupId = kafkaImportGroupId; + } + +} http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml b/extensions/router/router-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml new file mode 100644 index 0000000..4c36b9e --- /dev/null +++ b/extensions/router/router-core/src/main/resources/OSGI-INF/blueprint/blueprint.xml @@ -0,0 +1,94 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + ~ Licensed to the Apache Software Foundation (ASF) under one or more + ~ contributor license agreements. See the NOTICE file distributed with + ~ this work for additional information regarding copyright ownership. + ~ The ASF licenses this file to You under the Apache License, Version 2.0 + ~ (the "License"); you may not use this file except in compliance with + ~ the License. You may obtain a copy of the License at + ~ + ~ http://www.apache.org/licenses/LICENSE-2.0 + ~ + ~ Unless required by applicable law or agreed to in writing, software + ~ distributed under the License is distributed on an "AS IS" BASIS, + ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + ~ See the License for the specific language governing permissions and + ~ limitations under the License. + --> +<blueprint xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"; + xmlns:cm="http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.1.0"; + xmlns:camel="http://camel.apache.org/schema/blueprint"; + xsi:schemaLocation="http://www.osgi.org/xmlns/blueprint/v1.0.0 http://www.osgi.org/xmlns/blueprint/v1.0.0/blueprint.xsd + http://aries.apache.org/blueprint/xmlns/blueprint-cm/v1.1.0 http://aries.apache.org/schemas/blueprint-cm/blueprint-cm-1.1.0.xsd + http://camel.apache.org/schema/blueprint http://camel.apache.org/schema/blueprint/camel-blueprint.xsd";> + + <cm:property-placeholder persistent-id="org.apache.unomi.router" update-strategy="reload"> + <cm:default-properties> + <cm:property name="kafka.host" value="localhost"/> + <cm:property name="kafka.port" value="9092"/> + <cm:property name="kafka.import.topic" value="camel-deposit"/> + <cm:property name="kafka.import.groupId" value="unomi-import-group"/> + <cm:property name="import.oneshot.uploadDir" value="/tmp/oneshot_import_configs/"/> + </cm:default-properties> + </cm:property-placeholder> + + <bean id="unomiStorageProcessor" class="org.apache.unomi.router.core.processor.UnomiStorageProcessor"> + <property name="profileImportService" ref="profileImportService"/> + </bean> + + <bean id="importConfigByFileNameProcessor" class="org.apache.unomi.router.core.processor.ImportConfigByFileNameProcessor"> + <property name="importConfigurationService" ref="importConfigurationService"/> + </bean> + + <bean id="jacksonDataFormat" class="org.apache.camel.component.jackson.JacksonDataFormat"> + <property name="unmarshalType" value="org.apache.unomi.router.api.ProfileToImport"/> + </bean> + + <bean id="jacksonDataFormatImportConfig" class="org.apache.camel.model.dataformat.JsonDataFormat"> + <property name="unmarshalType" value="org.apache.unomi.router.api.ImportConfiguration"/> + <property name="library" value="Jackson"/> + </bean> + + <bean class="org.apache.camel.component.servlet.osgi.OsgiServletRegisterer" + init-method="register" + destroy-method="unregister"> + <property name="alias" value="/importConfigAdmin"/> + <property name="httpService" ref="httpService"/> + <property name="servlet" ref="camelServlet"/> + </bean> + + <bean id="camelServlet" class="org.apache.camel.component.servlet.CamelHttpTransportServlet"/> + + + <bean id="camelContext" class="org.apache.unomi.router.core.context.ProfileImportCamelContext" + init-method="initCamelContext" destroy-method="preDestroy"> + <property name="kafkaProps"> + <map> + <entry key="kafkaHost" value="${kafka.host}"/> + <entry key="kafkaPort" value="${kafka.port}"/> + <entry key="kafkaImportTopic" value="${kafka.import.topic}"/> + <entry key="kafkaImportGroupId" value="${kafka.import.groupId}"/> + </map> + </property> + <property name="uploadDir" value="${import.oneshot.uploadDir}"/> + <property name="unomiStorageProcessor" ref="unomiStorageProcessor"/> + <property name="importConfigByFileNameProcessor" ref="importConfigByFileNameProcessor"/> + <property name="importConfigurationService" ref="importConfigurationService"/> + <property name="jacksonDataFormat" ref="jacksonDataFormat"/> + <property name="bundleContext" ref="blueprintBundleContext"/> + </bean> + + <camel:camelContext id="httpEndpoint" xmlns="http://camel.apache.org/schema/blueprint";> + <camel:routeBuilder ref="profileImportConfigUpdateRouteBuilder" /> + </camel:camelContext> + + <bean id="profileImportConfigUpdateRouteBuilder" class="org.apache.unomi.router.core.route.ProfileImportConfigUpdateRouteBuilder"> + <property name="profileImportCamelContext" ref="camelContext"/> + </bean> + + <reference id="httpService" interface="org.osgi.service.http.HttpService"/> + <reference id="profileImportService" interface="org.apache.unomi.router.api.services.ProfileImportService"/> + <reference id="importConfigurationService" interface="org.apache.unomi.router.api.services.ImportConfigurationService"/> + +</blueprint> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-unomi/blob/89d4c8eb/extensions/router/router-core/src/main/resources/org.apache.unomi.router.cfg ---------------------------------------------------------------------- diff --git a/extensions/router/router-core/src/main/resources/org.apache.unomi.router.cfg b/extensions/router/router-core/src/main/resources/org.apache.unomi.router.cfg new file mode 100644 index 0000000..ff2c8ef --- /dev/null +++ b/extensions/router/router-core/src/main/resources/org.apache.unomi.router.cfg @@ -0,0 +1,25 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +#Kafka + settingskafka.host=localhost +kafka.port=9092 +kafka.import.topic=camel-deposit +kafka.import.groupId=unomi-import-group + +#Import One Shot upload directory +import.oneshot.uploadDir=/tmp/unomi_oneshot_import_configs/ \ No newline at end of file
