Initial check in accordance with Parallel IP

hawkbit-cache-redis/src/main/java/org/eclipse/hawkbit/cache/RedisConfiguration.java

@@ -0,0 +1,109 @@
+/**
+ * Copyright (c) 2015 Bosch Software Innovations GmbH and others.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.eclipse.hawkbit.cache;
+
+import org.eclipse.hawkbit.cache.eventbus.EventDistributor;
+import org.eclipse.hawkbit.tenancy.TenantAware;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cache.CacheManager;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.cache.RedisCacheManager;
+import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
+
+/**
+ * The spring Redis configuration which is enabled by using the profile
+ * {@code redis} to use a Redis server as cache.
+ *
+ *
+ *
+ */
+@Configuration
+@EnableConfigurationProperties(RedisProperties.class)
+public class RedisConfiguration {
+
+    @Autowired
+    private RedisProperties redisProperties;
+
+    @Autowired
+    private TenantAware tenantAware;
+
+    /**
+     * @return the {@link EventDistributor} to distribute and consume the events
+     *         from Redis
+     */
+    @Bean
+    public EventDistributor eventDistributor() {
+        return new EventDistributor();
+    }
+
+    /**
+     * @return the spring redis cache manager.
+     */
+    @Bean
+    public CacheManager cacheManager() {
+        return new TenantAwareCacheManager(new RedisCacheManager(redisTemplate()), tenantAware);
+    }
+
+    /**
+     * @return the redis connection factory to create a redis connection based
+     *         on the {@link RedisProperties}
+     */
+    @Bean
+    public JedisConnectionFactory jedisConnectionFactory() {
+        final JedisConnectionFactory factory = new JedisConnectionFactory();
+        factory.setHostName(redisProperties.getHost());
+        factory.setPort(redisProperties.getPort());
+        factory.setUsePool(true);
+        return factory;
+    }
+
+    /**
+     * @return the spring {@link RedisTemplate} configured with the necessary
+     *         object serializers
+     */
+    @Bean
+    public RedisTemplate<String, Object> redisTemplate() {
+        final RedisTemplate<String, Object> redisTemplate = new RedisTemplate<String, Object>();
+        redisTemplate.setConnectionFactory(jedisConnectionFactory());
+        redisTemplate.setKeySerializer(new JdkSerializationRedisSerializer());
+        redisTemplate.setHashValueSerializer(new JdkSerializationRedisSerializer());
+        redisTemplate.setValueSerializer(new JdkSerializationRedisSerializer());
+        return redisTemplate;
+    }
+
+    /**
+     * @return the spring-redis message listener adapter to consume messages
+     *         from the Redis server
+     */
+    @Bean
+    public MessageListenerAdapter messageListenerAdapter() {
+        final MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(eventDistributor());
+        messageListenerAdapter.setSerializer(new JdkSerializationRedisSerializer());
+        return messageListenerAdapter;
+    }
+
+    /**
+     * @return the spring-redis message listener container to register the
+     *         message listener adapter
+     */
+    @Bean
+    public RedisMessageListenerContainer redisContainer() {
+        final RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+        container.setConnectionFactory(jedisConnectionFactory());
+        container.addMessageListener(messageListenerAdapter(), eventDistributor().getTopics());
+        return container;
+    }
+
+}

hawkbit-cache-redis/src/main/java/org/eclipse/hawkbit/cache/RedisProperties.java

@@ -0,0 +1,56 @@
+/**
+ * Copyright (c) 2015 Bosch Software Innovations GmbH and others.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.eclipse.hawkbit.cache;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+/**
+ * Bean which holds the necessary properties for configuring the Redis
+ * connection.
+ *
+ *
+ *
+ *
+ */
+@ConfigurationProperties("hawkbit.server.redis")
+public class RedisProperties {
+
+    private String host;
+    private int port;
+
+    /**
+     * @return the host
+     */
+    public String getHost() {
+        return host;
+    }
+
+    /**
+     * @param host
+     *            the host to set
+     */
+    public void setHost(final String host) {
+        this.host = host;
+    }
+
+    /**
+     * @return the port
+     */
+    public int getPort() {
+        return port;
+    }
+
+    /**
+     * @param port
+     *            the port to set
+     */
+    public void setPort(final int port) {
+        this.port = port;
+    }
+}

hawkbit-cache-redis/src/main/java/org/eclipse/hawkbit/cache/annotation/EnableRedis.java

@@ -0,0 +1,29 @@
+/**
+ * Copyright (c) 2015 Bosch Software Innovations GmbH and others.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.eclipse.hawkbit.cache.annotation;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.eclipse.hawkbit.cache.RedisConfiguration;
+import org.springframework.context.annotation.Import;
+
+/**
+ * Annotation to enable redis caching.
+ * 
+ *
+ */
+@Target(ElementType.TYPE)
+@Retention(RetentionPolicy.RUNTIME)
+@Import(RedisConfiguration.class)
+public @interface EnableRedis {
+
+}

hawkbit-cache-redis/src/main/java/org/eclipse/hawkbit/cache/eventbus/EventDistributor.java

@@ -0,0 +1,138 @@
+/**
+ * Copyright (c) 2015 Bosch Software Innovations GmbH and others.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.eclipse.hawkbit.cache.eventbus;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.UUID;
+
+import org.eclipse.hawkbit.eventbus.EventSubscriber;
+import org.eclipse.hawkbit.eventbus.event.DistributedEvent;
+import org.eclipse.hawkbit.eventbus.event.Event;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.PatternTopic;
+import org.springframework.data.redis.listener.Topic;
+import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
+
+import com.google.common.eventbus.EventBus;
+import com.google.common.eventbus.Subscribe;
+
+/**
+ *
+ *
+ */
+@EventSubscriber
+public class EventDistributor {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(EventDistributor.class);
+    /**
+     * the node id to filter self published events in the redis message
+     * subscriber.
+     */
+    private static final String NODE_ID = UUID.randomUUID().toString();
+
+    private static final String DISTRIBUTION_CHANNEL_TOPIC = "com/bosch/sp/distEvent";
+    private static final String SEND_DISTRIBUTION_CHANNEL = DISTRIBUTION_CHANNEL_TOPIC + "/" + NODE_ID;
+    private static final String SUB_DISTRIBUTION_CHANNEL = DISTRIBUTION_CHANNEL_TOPIC + "*";
+
+    @Autowired(required = false)
+    private RedisTemplate<String, Object> redisTemplate;
+
+    @Autowired
+    private EventBus eventBus;
+
+    /**
+     * consumes all {@link DistributedEvent}s posted on the {@link EventBus} and
+     * distribute them to the Redis server.
+     * 
+     * @param event
+     *            the distributed event posted on the {@link EventBus}
+     */
+    @Subscribe
+    public void distribute(final DistributedEvent event) {
+        if (redisTemplate != null) {
+            if (!NODE_ID.equals(event.getNodeId())) {
+                logDistributingEvent(event, SEND_DISTRIBUTION_CHANNEL);
+                event.setOriginNodeId(NODE_ID);
+                redisTemplate.convertAndSend(SEND_DISTRIBUTION_CHANNEL, event);
+            }
+        } else {
+            logNotDistributingEvent(event, SEND_DISTRIBUTION_CHANNEL);
+        }
+    }
+
+    /**
+     * message listener callback method for the {@link MessageListenerAdapter}
+     * which calls the method in case a message is received from the Redis
+     * server of the type {@link DistributedEventWrapper}.
+     * 
+     * @param event
+     *            the {@link DistributedEventWrapper} event which was received
+     *            by the Redis client
+     * @param channel
+     *            the on which the event was received
+     */
+    public void handleMessage(final DistributedEvent event, final String channel) {
+        LOGGER.trace("retrieving event from redis {} on channel {}, posting to the local event bus", event, channel);
+        if (!NODE_ID.equals(event.getOriginNodeId())) {
+            event.setNodeId(NODE_ID);
+            eventBus.post(event);
+        }
+    }
+
+    /**
+     * @return a collection of all topics which this Redis message listener
+     *         wants to subscribe
+     */
+    public Collection<Topic> getTopics() {
+        final List<Topic> topics = new ArrayList<Topic>();
+        topics.add(new PatternTopic(SUB_DISTRIBUTION_CHANNEL));
+        return topics;
+    }
+
+    private void logDistributingEvent(final Event event, final String channel) {
+        LOGGER.trace("distributing event {} from node {} to topic {}", event, NODE_ID, channel);
+    }
+
+    private void logNotDistributingEvent(final Event event, final String channel) {
+        LOGGER.debug("no redis template configured, event {} will not be distributed to channel {} from node {}", event,
+                channel, NODE_ID);
+    }
+
+    /**
+     * testing purposes.
+     * 
+     * @param redisTemplate
+     *            the redisTemplate to set
+     */
+    void setRedisTemplate(final RedisTemplate<String, Object> redisTemplate) {
+        this.redisTemplate = redisTemplate;
+    }
+
+    /**
+     * testing purposes.
+     * 
+     * @param eventBus
+     *            the eventBus to set
+     */
+    void setEventBus(final EventBus eventBus) {
+        this.eventBus = eventBus;
+    }
+
+    /**
+     * @return the nodeId
+     */
+    static String getNodeId() {
+        return NODE_ID;
+    }
+}

hawkbit-cache-redis/src/test/java/org/eclipse/hawkbit/cache/eventbus/EventDistributorTest.java

@@ -0,0 +1,222 @@
+/**
+ * Copyright (c) 2015 Bosch Software Innovations GmbH and others.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.eclipse.hawkbit.cache.eventbus;
+
+import static org.fest.assertions.api.Assertions.assertThat;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.util.Collection;
+
+import org.eclipse.hawkbit.eventbus.event.DownloadProgressEvent;
+import org.eclipse.hawkbit.eventbus.event.EntityEvent;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.Topic;
+import org.springframework.hateoas.Identifiable;
+
+import com.google.common.eventbus.EventBus;
+
+@RunWith(MockitoJUnitRunner.class)
+public class EventDistributorTest {
+
+    @Mock
+    private RedisTemplate<String, Object> redisTemplateMock;
+
+    @Mock
+    private EventBus eventBusMock;
+
+    private EventDistributor underTest;
+
+    @Before
+    public void before() {
+        underTest = new EventDistributor();
+        underTest.setEventBus(eventBusMock);
+        underTest.setRedisTemplate(redisTemplateMock);
+    }
+
+    @Test
+    public void distributeDistributedEventSendsToRedis() {
+
+        final DownloadProgressEvent event = new DownloadProgressEvent("tenant", 123L, 10);
+        underTest.distribute(event);
+
+        // origin node ID should be set by distributing the event
+        assertThat(event.getOriginNodeId()).isNotNull();
+        verify(redisTemplateMock).convertAndSend(anyString(), eq(event));
+    }
+
+    @Test
+    public void dontDistributeDistributedEventIfSameNode() {
+        final String knownNodeId = EventDistributor.getNodeId();
+        final DownloadProgressEvent event = new DownloadProgressEvent("tenant", 123L, 10);
+        event.setNodeId(knownNodeId);
+
+        // test
+        underTest.distribute(event);
+
+        assertThat(event.getOriginNodeId()).isNull();
+        verify(redisTemplateMock, times(0)).convertAndSend(anyString(), eq(event));
+    }
+
+    @Test
+    public void handleDistributedMessageFromRedis() {
+        final DownloadProgressEvent event = new DownloadProgressEvent("tenant", 123L, 10);
+        final String knownChannel = "someChannel";
+
+        underTest.handleMessage(event, knownChannel);
+
+        assertThat(event.getNodeId()).isEqualTo(EventDistributor.getNodeId());
+        verify(eventBusMock).post(eq(event));
+    }
+
+    @Test
+    public void handleDistributedMessageFilteredIfSameNodeId() {
+        final DownloadProgressEvent event = new DownloadProgressEvent("tenant", 123L, 10);
+        final String knownChannel = "someChannel";
+        event.setOriginNodeId(EventDistributor.getNodeId());
+
+        underTest.handleMessage(event, knownChannel);
+
+        assertThat(event.getNodeId()).isNull();
+        verify(eventBusMock, times(0)).post(eq(event));
+    }
+
+    @Test
+    public void subscribedTopicsContains2PatternTopics() {
+        final Collection<Topic> topics = underTest.getTopics();
+        assertThat(topics).hasSize(1);
+    }
+
+    private class TestEntityEvent implements EntityEvent {
+
+        private String originNodeId;
+        private String nodeId;
+        private final String tenant;
+        private final MyEntity entity;
+
+        /**
+         * @param myEntity
+         */
+        public TestEntityEvent(final MyEntity myEntity, final String tenant) {
+            this.entity = myEntity;
+            this.tenant = tenant;
+        }
+
+        /*
+         * (non-Javadoc)
+         *
+         * @see org.eclipse.hawkbit.server.eventbus.event.Event#getRevision()
+         */
+        @Override
+        public long getRevision() {
+            return 0;
+        }
+
+        /*
+         * (non-Javadoc)
+         *
+         * @see
+         * org.eclipse.hawkbit.server.eventbus.event.NodeAware#getOriginNodeId()
+         */
+        @Override
+        public String getOriginNodeId() {
+            return originNodeId;
+        }
+
+        /*
+         * (non-Javadoc)
+         *
+         * @see
+         * org.eclipse.hawkbit.server.eventbus.event.NodeAware#setOriginNodeId(
+         * java. lang.String)
+         */
+        @Override
+        public void setOriginNodeId(final String originNodeId) {
+            this.originNodeId = originNodeId;
+        }
+
+        /*
+         * (non-Javadoc)
+         *
+         * @see org.eclipse.hawkbit.server.eventbus.event.NodeAware#getNodeId()
+         */
+        @Override
+        public String getNodeId() {
+            return nodeId;
+        }
+
+        /*
+         * (non-Javadoc)
+         *
+         * @see
+         * org.eclipse.hawkbit.server.eventbus.event.NodeAware#setNodeId(java.
+         * lang. String)
+         */
+        @Override
+        public void setNodeId(final String nodeId) {
+            this.nodeId = nodeId;
+        }
+
+        /*
+         * (non-Javadoc)
+         *
+         * @see
+         * org.eclipse.hawkbit.server.eventbus.event.EntityEvent#getEntity(java.
+         * lang. Class)
+         */
+        @Override
+        public <E> E getEntity(final Class<E> entityClass) {
+            return entityClass.cast(entity);
+        }
+
+        /*
+         * (non-Javadoc)
+         *
+         * @see
+         * org.eclipse.hawkbit.server.eventbus.event.EntityEvent#getEntity()
+         */
+        @Override
+        public Object getEntity() {
+            return entity;
+        }
+
+        /*
+         * (non-Javadoc)
+         *
+         * @see
+         * org.eclipse.hawkbit.server.eventbus.event.EntityEvent#getTenant()
+         */
+        @Override
+        public String getTenant() {
+            return tenant;
+        }
+    }
+
+    private class MyEntity implements Identifiable<String> {
+        private final String id = "123";
+
+        /*
+         * (non-Javadoc)
+         *
+         * @see org.springframework.hateoas.Identifiable#getId()
+         */
+        @Override
+        public String getId() {
+            return id;
+        }
+    }
+
+}

hawkbit-cache-redis/src/test/java/org/eclipse/hawkbit/cache/redis/RedisPropertiesTest.java

@@ -0,0 +1,31 @@
+/**
+ * Copyright (c) 2015 Bosch Software Innovations GmbH and others.
+ *
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ */
+package org.eclipse.hawkbit.cache.redis;
+
+import static org.fest.assertions.api.Assertions.assertThat;
+
+import org.eclipse.hawkbit.cache.RedisProperties;
+import org.junit.Test;
+
+public class RedisPropertiesTest {
+
+    @Test
+    public void setAndGetProps() {
+        final String knownHost = "bumlux";
+        final int knownPort = 1234;
+
+        final RedisProperties underTest = new RedisProperties();
+        underTest.setHost(knownHost);
+        underTest.setPort(knownPort);
+
+        assertThat(underTest.getHost()).isEqualTo(knownHost);
+        assertThat(underTest.getPort()).isEqualTo(knownPort);
+    }
+
+}