Redis Cache, Message Queue example

RedisCacheServiceApplication.java

package com.spdeepak.redis;

import com.spdeepak.redis.entity.LineItem;
import com.spdeepak.redis.entity.Order;
import com.spdeepak.redis.repository.LineItemRepository;
import com.spdeepak.redis.repository.OrderRepository;
import com.spdeepak.redis.service.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cache.CacheManager;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.data.geo.Circle;
import org.springframework.data.geo.Distance;
import org.springframework.data.geo.GeoResults;
import org.springframework.data.geo.Point;
import org.springframework.data.redis.cache.RedisCacheManager;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisGeoCommands;
import org.springframework.data.redis.core.GeoOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.time.Instant;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.Random;

@Slf4j
@EnableCaching
@SpringBootApplication
public class RedisCacheServiceApplication {

	private String topic = "chat";

	public static void main(String[] args) {
		SpringApplication.run(RedisCacheServiceApplication.class, args);
	}

	@Bean
	ApplicationRunner geography(RedisTemplate<String, String> redisTemplate) {
		return titleRunner("geography", args -> {
			GeoOperations<String, String> geo = redisTemplate.opsForGeo();
			geo.add("Sicily", new Point(13.361389, 38.115556), "Arigento");
			geo.add("Sicily", new Point(15.087269, 37.502669), "Catania");
			geo.add("Sicily", new Point(13.583333, 37.316667), "Palermo");

			Circle circle = new Circle(new Point(13.583333, 37.316667), new Distance(100, RedisGeoCommands.DistanceUnit.KILOMETERS));
			GeoResults<RedisGeoCommands.GeoLocation<String>> geoResults = geo.radius("Sicily", circle);
			geoResults.getContent()
					.forEach(c -> log.info(c.toString()));
		});
	}

	private ApplicationRunner titleRunner(String title, ApplicationRunner runner) {
		return args -> {
			log.info("Title ====== {} :", title.toUpperCase());
			runner.run(args);
		};
	}

	@Bean
	ApplicationRunner repositories(LineItemRepository lineItemRepository, OrderRepository orderRepository) {
		return titleRunner("repositories", args -> {
			Long orderId = generateId();
			List<LineItem> lineItems = Arrays.asList(new LineItem(orderId, generateId(), "plunger"), new LineItem(orderId, generateId(), "soup"), new LineItem(orderId, generateId(), "coffee mug"));
			lineItems.stream()
					.map(lineItemRepository::save)
					.forEach(li -> log.info(li.toString()));

			Order order = new Order(orderId, new Date(), lineItems);
			orderRepository.save(order);
			Collection<Order> found = orderRepository.findByWhen(order.getWhen());
			found.forEach(o -> log.info("Found: {}", found.toString()));
		});
	}

	private Long generateId() {
		long tmp = new Random().nextLong();
		return Math.max(tmp, tmp * -1);
	}

	/**
	 * This is how we do a message publish and subscribe the beans below
	 *
	 * @param redisTemplate
	 * @return
	 */
	@Bean
	ApplicationRunner pubSub(RedisTemplate<String, String> redisTemplate) {
		return titleRunner("publish/scubscribe", args -> {
			log.info("Sending Message to Topic: {}", topic);
			redisTemplate.convertAndSend(topic, "Hello, world @" + Instant.now()
					.toString());
			log.info("Message sent to Topic: {}", topic);
		});
	}

	/**
	 * This is how we listen to messages
	 *
	 * @param redisConnectionFactory
	 * @return
	 */
	@Bean
	RedisMessageListenerContainer redisMessageListener(RedisConnectionFactory redisConnectionFactory) {
		MessageListener messageListener = (message, pattern) -> {
			String str = new String(message.getBody());
			log.info("Message from {}:{}", topic, str);
		};
		RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
		redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
		redisMessageListenerContainer.addMessageListener(messageListener, new PatternTopic(this.topic));
		log.info("Added message listener for Topic: {}", topic);
		return redisMessageListenerContainer;
	}

	@Bean
	CacheManager redisCache(RedisConnectionFactory redisConnectionFactory) {
		return RedisCacheManager.builder(redisConnectionFactory)
				.build();
	}

	@Bean
	ApplicationRunner cache(OrderService orderService) {
		return titleRunner("caching", args -> {
			Runnable measure = () -> {
				Optional<Order> order = orderService.findById(1L);
				log.info("Order: {}", order.get());
			};
			log.info("First: {}", measure(measure));
			log.info("Second: {}", measure(measure));
			log.info("Third: {}", measure(measure));
		});
	}

	private long measure(Runnable runnable) {
		long start = System.currentTimeMillis();
		runnable.run();
		long stop = System.currentTimeMillis();
		return stop - start;
	}

}