Merge pull request #1117 from agulenko/master

KAA-1423: Added Endpoint Specific Configuration Support
# Conflicts:
#	server/common/admin-rest-client/src/main/java/org/kaaproject/kaa/server/common/admin/AdminClient.java
#	server/common/dao/src/main/java/org/kaaproject/kaa/server/common/dao/ConfigurationService.java
#	server/common/dao/src/main/java/org/kaaproject/kaa/server/common/dao/service/ConfigurationServiceImpl.java
#	server/common/dao/src/main/java/org/kaaproject/kaa/server/common/dao/service/UserConfigurationServiceImpl.java
#	server/common/dao/src/main/java/org/kaaproject/kaa/server/common/dao/service/Validator.java
#	server/common/dao/src/test/java/org/kaaproject/kaa/server/common/dao/AbstractTest.java
#	server/common/dto/src/main/java/org/kaaproject/kaa/common/dto/EndpointProfileDto.java
#	server/common/nosql/cassandra-dao/src/main/java/org/kaaproject/kaa/server/common/nosql/cassandra/dao/model/CassandraEndpointProfile.java
#	server/common/nosql/cassandra-dao/src/main/java/org/kaaproject/kaa/server/common/nosql/cassandra/dao/model/CassandraModelConstants.java
#	server/common/nosql/cassandra-dao/src/test/java/org/kaaproject/kaa/server/common/nosql/cassandra/dao/AbstractCassandraTest.java
#	server/common/nosql/mongo-dao/src/main/java/org/kaaproject/kaa/server/common/nosql/mongo/dao/model/MongoEndpointProfile.java
#	server/common/nosql/mongo-dao/src/main/java/org/kaaproject/kaa/server/common/nosql/mongo/dao/model/MongoModelConstants.java
#	server/common/nosql/mongo-dao/src/test/java/org/kaaproject/kaa/server/common/nosql/mongo/dao/AbstractMongoTest.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/bootstrap/ThriftOperationsServer.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/cli/CliThriftException.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/cli/CliThriftService.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/cli/MemoryUsage.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/operations/EndpointEvent.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/operations/EndpointRouteUpdate.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/operations/EventClassFamilyVersion.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/operations/Message.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/operations/Notification.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/operations/OperationsThriftService.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/operations/RedirectionRule.java
#	server/common/thrift/src/main/thrift-java/org/kaaproject/kaa/server/common/thrift/gen/operations/UserConfigurationUpdate.java
#	server/node/src/main/java/org/kaaproject/kaa/server/admin/controller/ConfigurationController.java
#	server/node/src/main/java/org/kaaproject/kaa/server/admin/services/AbstractAdminService.java
#	server/node/src/main/java/org/kaaproject/kaa/server/admin/services/ConfigurationServiceImpl.java
#	server/node/src/main/java/org/kaaproject/kaa/server/admin/services/util/Utils.java
#	server/node/src/main/java/org/kaaproject/kaa/server/admin/shared/services/ConfigurationService.java
#	server/node/src/main/java/org/kaaproject/kaa/server/control/service/ControlService.java
#	server/node/src/main/java/org/kaaproject/kaa/server/control/service/DefaultControlService.java
#	server/node/src/main/java/org/kaaproject/kaa/server/node/service/thrift/OperationsServiceMsg.java
#	server/node/src/main/java/org/kaaproject/kaa/server/operations/service/DefaultOperationsService.java
#	server/node/src/main/java/org/kaaproject/kaa/server/operations/service/OperationsService.java
#	server/node/src/main/java/org/kaaproject/kaa/server/operations/service/akka/actors/core/endpoint/global/GlobalEndpointActorMessageProcessor.java
#	server/node/src/main/java/org/kaaproject/kaa/server/operations/service/akka/actors/core/endpoint/local/LocalEndpointActorMessageProcessor.java
#	server/node/src/main/java/org/kaaproject/kaa/server/operations/service/akka/actors/core/endpoint/local/LocalEndpointActorState.java
#	server/node/src/main/java/org/kaaproject/kaa/server/operations/service/cache/ConfigurationCacheEntry.java
#	server/node/src/main/java/org/kaaproject/kaa/server/operations/service/cache/DeltaCacheKey.java
#	server/node/src/main/java/org/kaaproject/kaa/server/operations/service/cluster/ClusterService.java
#	server/node/src/main/java/org/kaaproject/kaa/server/operations/service/cluster/DefaultClusterService.java
#	server/node/src/main/java/org/kaaproject/kaa/server/operations/service/delta/DefaultDeltaService.java
#	server/node/src/main/java/org/kaaproject/kaa/server/operations/service/thrift/OperationsThriftServiceImpl.java
#	server/node/src/test/java/org/kaaproject/kaa/server/operations/service/OperationsServiceIT.java
#	server/node/src/test/java/org/kaaproject/kaa/server/operations/service/akka/DefaultAkkaServiceTest.java
#	server/node/src/test/java/org/kaaproject/kaa/server/operations/service/cache/DeltaCacheKeyTest.java
#	server/node/src/test/java/org/kaaproject/kaa/server/operations/service/delta/DeltaServiceIT.java
#	server/node/src/test/java/org/kaaproject/kaa/server/operations/service/event/ESTestOperationsService.java
#	server/upgrade/data-migration-0.9.0-0.10.0/src/main/java/org/kaaproject/kaa/server/datamigration/MigrateData.java

server/common/admin-rest-client/src/main/java/org/kaaproject/kaa/server/common/admin/AdminClient.java

@@ -35,6 +35,7 @@
 import org.kaaproject.kaa.common.dto.EndpointProfileSchemaDto;
 import org.kaaproject.kaa.common.dto.EndpointProfilesBodyDto;
 import org.kaaproject.kaa.common.dto.EndpointProfilesPageDto;
+import org.kaaproject.kaa.common.dto.EndpointSpecificConfigurationDto;
 import org.kaaproject.kaa.common.dto.EndpointUserConfigurationDto;
 import org.kaaproject.kaa.common.dto.NotificationDto;
 import org.kaaproject.kaa.common.dto.NotificationSchemaDto;
@@ -843,6 +844,46 @@ public void editUserConfiguration(EndpointUserConfigurationDto endpointUserConfi
         endpointUserConfiguration);
   }
 
+  public EndpointSpecificConfigurationDto editEndpointSpecificConfiguration(EndpointSpecificConfigurationDto configuration) throws Exception {
+    return restTemplate.postForObject(restTemplate.getUrl() + "endpointSpecificConfiguration", configuration, EndpointSpecificConfigurationDto.class);
+  }
+
+  public void deleteActiveEndpointSpecificConfiguration(String endpointKeyHash) throws Exception {
+    restTemplate.delete(restTemplate.getUrl() + "endpointSpecificConfiguration/{endpointKeyHash}", toUrlSafe(endpointKeyHash));
+  }
+
+  public EndpointSpecificConfigurationDto findActiveEndpointSpecificConfiguration(String endpointKeyHash) throws Exception {
+    return restTemplate.getForObject(restTemplate.getUrl() + "endpointSpecificConfiguration/{endpointKeyHash}",
+        EndpointSpecificConfigurationDto.class, toUrlSafe(endpointKeyHash));
+  }
+
+  /**
+   * Delete endpoint specific configuration.
+   *
+   * @param endpointKeyHash endpoint key hash
+   * @param configurationSchemaVersion configuration schema version
+   */
+  public void deleteEndpointSpecificConfigurationByEndpointKeyHashAndConfigurationSchemaVersion(String endpointKeyHash,
+                                                                                                int configurationSchemaVersion) throws Exception {
+    restTemplate.delete(restTemplate.getUrl()
+            + "endpointSpecificConfiguration/{endpointKeyHash}?configurationSchemaVersion={configurationSchemaVersion}",
+        toUrlSafe(endpointKeyHash), configurationSchemaVersion);
+  }
+
+  /**
+   * Find endpoint specific configuration.
+   *
+   * @param endpointKeyHash endpoint key hash
+   * @param configurationSchemaVersion configuration schema version
+   */
+  public EndpointSpecificConfigurationDto findEndpointSpecificConfigurationByEndpointKeyHashAndConfigurationSchemaVersion(String endpointKeyHash,
+                                                                                                                          int configurationSchemaVersion)
+      throws Exception {
+    return restTemplate.getForObject(restTemplate.getUrl()
+            + "endpointSpecificConfiguration/{endpointKeyHash}?configurationSchemaVersion={configurationSchemaVersion}",
+        EndpointSpecificConfigurationDto.class, toUrlSafe(endpointKeyHash), configurationSchemaVersion);
+  }
+
   public ProfileFilterDto editProfileFilter(ProfileFilterDto profileFilter) throws Exception {
     return restTemplate.postForObject(restTemplate.getUrl() + "profileFilter",
         profileFilter, ProfileFilterDto.class);

server/common/dao/src/main/java/org/kaaproject/kaa/server/common/dao/ConfigurationService.java

@@ -204,6 +204,15 @@ ConfigurationSchemaDto saveConfSchema(ConfigurationSchemaDto configurationSchema
    */
   ConfigurationSchemaDto findConfSchemaById(String id);
 
+  /**
+   * Normalize configuration according to override configuration schema.
+   *
+   * @param appId             application ID
+   * @param schemaVersion     configuration schema version
+   * @param configurationBody configuration to validate
+   */
+  String normalizeAccordingToOverrideConfigurationSchema(String appId, int schemaVersion, String configurationBody);
+
   /**
    * Remove configuration schemas by application id.
    *

server/common/dao/src/main/java/org/kaaproject/kaa/server/common/dao/EndpointSpecificConfigurationService.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2014-2016 CyberVision, Inc.
+ *
+ * Licensed 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.kaaproject.kaa.server.common.dao;
+
+import org.kaaproject.kaa.common.dto.EndpointProfileDto;
+import org.kaaproject.kaa.common.dto.EndpointSpecificConfigurationDto;
+
+import java.util.Optional;
+
+/**
+ * The interface Endpoint specific configuration service.
+ */
+public interface EndpointSpecificConfigurationService {
+
+  /**
+   * Find currently active endpoint specific configuration by endpoint key hash.
+   *
+   * @param endpointKeyHash endpoint key hash
+   * @return the endpoint specific configuration
+   */
+  Optional<EndpointSpecificConfigurationDto> findActiveConfigurationByEndpointKeyHash(byte[] endpointKeyHash);
+
+  /**
+   * Find endpoint specific configuration by endpoint key hash and configuration schema version.
+   *
+   * @param endpointKeyHash   endpoint key hash
+   * @param confSchemaVersion configuration schema version
+   * @return the endpoint specific configuration
+   */
+  Optional<EndpointSpecificConfigurationDto> findByEndpointKeyHashAndConfSchemaVersion(byte[] endpointKeyHash, Integer confSchemaVersion);
+
+  /**
+   * Find currently active endpoint specific configuration by endpoint profile.
+   *
+   * @param endpointProfileDto endpoint key hash
+   * @return the endpoint specific configuration
+   */
+  Optional<EndpointSpecificConfigurationDto> findActiveConfigurationByEndpointProfile(EndpointProfileDto endpointProfileDto);
+
+  /**
+   * Delete currently active endpoint specific configuration by endpoint key hash.
+   *
+   * @param endpointKeyHash endpoint key hash
+   * @return deleted endpoint specific configuration
+   */
+  Optional<EndpointSpecificConfigurationDto> deleteActiveConfigurationByEndpointKeyHash(byte[] endpointKeyHash);
+
+  /**
+   * Delete endpoint specific configuration by endpoint key hash and configuration schema version.
+   *
+   * @param endpointKeyHash   endpoint key hash
+   * @param confSchemaVersion configuration schema version
+   * @return deleted endpoint specific configuration
+   */
+  Optional<EndpointSpecificConfigurationDto> deleteByEndpointKeyHashAndConfSchemaVersion(byte[] endpointKeyHash, Integer confSchemaVersion);
+
+  /**
+   * Save endpoint specific configuration.
+   *
+   * @param configurationDto endpoint specific configuration
+   * @return saved endpoint specific configuration
+   */
+  EndpointSpecificConfigurationDto save(EndpointSpecificConfigurationDto configurationDto);
+
+}

server/common/dao/src/main/java/org/kaaproject/kaa/server/common/dao/impl/EndpointSpecificConfigurationDao.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2014-2016 CyberVision, Inc.
+ *
+ * Licensed 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.kaaproject.kaa.server.common.dao.impl;
+
+import org.kaaproject.kaa.common.dto.EndpointSpecificConfigurationDto;
+import org.kaaproject.kaa.server.common.dao.model.EndpointSpecificConfiguration;
+
+public interface EndpointSpecificConfigurationDao<T extends EndpointSpecificConfiguration> extends Dao<T, String> {
+
+  void removeByEndpointKeyHashAndConfigurationVersion(byte[] endpointKeyHash, Integer confSchemaVersion);
+
+  EndpointSpecificConfiguration findByEndpointKeyHashAndConfigurationVersion(byte[] endpointKeyHash, int configurationVersion);
+
+  EndpointSpecificConfiguration save(EndpointSpecificConfigurationDto endpointSpecificConfigurationDto);
+
+}

server/common/dao/src/main/java/org/kaaproject/kaa/server/common/dao/model/EndpointSpecificConfiguration.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2014-2016 CyberVision, Inc.
+ *
+ * Licensed 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.kaaproject.kaa.server.common.dao.model;
+
+import org.kaaproject.kaa.common.dto.EndpointSpecificConfigurationDto;
+import org.kaaproject.kaa.common.dto.HasVersion;
+
+import java.io.Serializable;
+
+/**
+ * An endpoint specific configuration provides single endpoint configuration without creating dedicated endpoint group.
+ * Endpoint specific configuration has highest priority in configuration override.
+ */
+public interface EndpointSpecificConfiguration extends ToDto<EndpointSpecificConfigurationDto>, HasVersion, Serializable {
+
+}

server/common/dao/src/main/java/org/kaaproject/kaa/server/common/dao/service/ConfigurationServiceImpl.java

@@ -53,6 +53,7 @@
 import org.kaaproject.kaa.server.common.core.configuration.BaseData;
 import org.kaaproject.kaa.server.common.core.configuration.BaseDataFactory;
 import org.kaaproject.kaa.server.common.core.configuration.KaaData;
+import org.kaaproject.kaa.server.common.core.configuration.OverrideData;
 import org.kaaproject.kaa.server.common.core.configuration.OverrideDataFactory;
 import org.kaaproject.kaa.server.common.core.schema.BaseSchema;
 import org.kaaproject.kaa.server.common.core.schema.DataSchema;
@@ -77,6 +78,7 @@
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -514,6 +516,34 @@ public ConfigurationSchemaDto findConfSchemaById(String id) {
     return getDto(configurationSchemaDao.findById(id));
   }
 
+  @Override
+  public String normalizeAccordingToOverrideConfigurationSchema(String appId, int schemaVersion, String configurationBody) {
+    ConfigurationSchemaDto schemaDto = this.findConfSchemaByAppIdAndVersion(appId, schemaVersion);
+    if (schemaDto != null) {
+      OverrideSchema overrideSchema = new OverrideSchema(schemaDto.getOverrideSchema());
+      LOG.debug("Create default UUID validator with override schema: {}", overrideSchema.getRawSchema());
+      UuidValidator<OverrideData> uuidValidator = new DefaultUuidValidator<>(overrideSchema, new OverrideDataFactory());
+      GenericAvroConverter<GenericRecord> avroConverter = new GenericAvroConverter<>(overrideSchema.getRawSchema());
+      try {
+        GenericRecord configRecord = avroConverter.decodeJson(configurationBody);
+        // TODO: Need to use last active configuration instead of null. Will be changed after supporting delta configuration
+        KaaData<OverrideSchema> body = uuidValidator.validateUuidFields(configRecord, null);
+        if (body != null) {
+          return body.getRawData();
+        } else {
+          LOG.warn("Validated configuration body is empty");
+          throw new IncorrectParameterException("Validated configuration body is empty");
+        }
+      } catch (IOException ex) {
+        LOG.error("Invalid configuration for override schema.", ex);
+        throw new IncorrectParameterException("Invalid configuration for override schema.");
+      }
+    } else {
+      LOG.warn("Can't find configuration schema with version {}.", schemaVersion);
+      throw new IncorrectParameterException("Can't find configuration schema for specified version.");
+    }
+  }
+
   @Override
   public void removeConfSchemasByAppId(String appId) {
     validateSqlId(appId, "Incorrect application id " + appId