import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
public void subscribe(ChannelHandlerContext context, String channel, String subId) {

        // dont want to trust the client to guive me a unique session id
        // so, going to add some salt to the session id and try to mitigate the chances for the client miss behaving
        String listenerId = Listener.getListenerUniqueId(context, subId);

        if (!subscriptions.containsKey(listenerId)) {
            Topic topic = new ChannelTopic(properties.getRedisChannelPrefix() + channel);
            LOGGER.debug("Subscribing client address={} subscription={} mapping channel={} onto topic={}",
                context.channel().remoteAddress(), subId, channel, topic);

            PerSubscriptionListenerAdapter listener = new PerSubscriptionListenerAdapter(context, subId, listenerId);
            redisMessageListenerContainer.addMessageListener(listener, topic);
            subscriptions.put(listenerId, listener);
        } else {

            LOGGER.warn(
                "Listner [{}] already exists. Maybe clients are not sending unique session ids.",
                listenerId);
        }
    }
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
@Bean
ChannelTopic topic() {
    return new ChannelTopic("pubsub:queue");
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
private void setupListener(RedisSerializer<?> listenerSerializer) throws InterruptedException {
	MessageListenerAdapter listener = new MessageListenerAdapter();
	listener.setDelegate(new Listener(latch));
	listener.setSerializer(listenerSerializer);
	listener.afterPropertiesSet();

	this.container = new RedisMessageListenerContainer();
	container.setConnectionFactory(connectionFactory);
	container.afterPropertiesSet();
	container.addMessageListener(listener, Collections.<Topic> singletonList(new ChannelTopic(TOPIC)));
	container.start();
	Thread.sleep(1000);
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
private Collection<Topic> toTopics(String channels) {
    String[] channelsArrays = channels.split(",");
    List<Topic> topics = new ArrayList<>();
    for (String channel : channelsArrays) {
        String name = channel.trim();
        if (Command.PSUBSCRIBE.equals(redisConfiguration.getCommand())) {
            topics.add(new PatternTopic(name));
        } else if (Command.SUBSCRIBE.equals(redisConfiguration.getCommand())) {
            topics.add(new ChannelTopic(name));
        } else {
            throw new IllegalArgumentException("Unsupported Command " + redisConfiguration.getCommand());
        }
    }
    return topics;
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
@Test
public void registerConsumerForTwoChannelTopics() throws Exception {
    ArgumentCaptor<Collection> collectionCaptor = ArgumentCaptor.forClass(Collection.class);
    verify(listenerContainer).addMessageListener(any(MessageListener.class), collectionCaptor.capture());

    Collection<ChannelTopic> topics = collectionCaptor.getValue();
    Iterator<ChannelTopic> topicIterator = topics.iterator();

    Topic firstTopic = topicIterator.next();
    Topic twoTopic = topicIterator.next();

    assertEquals("one", firstTopic.getTopic());
    assertEquals("two", twoTopic.getTopic());
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
public MessagePublisherImpl(final RedisTemplate<String, Object> redisTemplate, final ChannelTopic topic) {
    this.redisTemplate = redisTemplate;
    this.topic = topic;
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
@Bean
ChannelTopic topic() {
    return new ChannelTopic(topic);
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
public RedisInfoPublisher(StringRedisTemplate redisTemplate, ChannelTopic topic) {
    this.redisTemplate = redisTemplate;
    this.topic = topic;
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
@Override
public ChannelTopic getTopic(String channel) {
   return new ChannelTopic(keyString(CHANNEL, channel));
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
@Bean
public Topic gameTopic() {
    return new ChannelTopic("pubsub:game");
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
@Bean
public Topic chatTopic() {
    return new ChannelTopic("pubsub:chat");
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
/**
 * 채팅방 생성 로직 
 */
public void createRoom(CreateRoom req, String myId, ChannelHandlerContext ctx){
	Channel ch = ctx.channel();
	JSONObject result = null;
	try{
		//방정보 생성
		String roomId = indexDAO.increaseRoomIndex();	//방 인덱스
		List<String> idList = req.getIdList();
		if(idList==null)
			throw new Exception("초대할 친구 리스트가 비었습니다.");

		//방접속 정보 저장
		messageDAO.saveRoomList(myId, roomId);
		messageDAO.saveRoomUser(roomId, myId);
		MessageListener listener = new MessageListenerImpl(myId, roomId, failDAO);
		redisContainer.addMessageListener(listener, new ChannelTopic(roomId));	//리스너 자원 관리
		idChannelListenerMap.get(myId).put(roomId, listener);

		for(String id : idList){
			Member member = new Member();
			member.setId(id); 
			//존재하는 사용자 일 경우에만
			if(memberService.isExistsMember(member) == true && id.equals(myId) == false ){
				messageDAO.saveRoomList(id, roomId);
				messageDAO.saveRoomUser(roomId, id);
				TokenListItem item = memberService.getTokenListItem(id);
				String friendToken = item.getToken();
				MessageListener listenerForSubscriber = new MessageListenerImpl(id, roomId, failDAO);
				redisContainer.addMessageListener(listenerForSubscriber, new ChannelTopic(roomId));	
				idChannelListenerMap.get(id).put(roomId, listenerForSubscriber);
			}
		}

		//초기 메세지 저장/전송 (publish)

		String messageIndex = indexDAO.increaseMessageIndex(roomId);

		Packet packet = new Packet();
		packet.setFromId(myId);
		packet.setContent(myId+"님의 채팅방에 초대 되었습니다.");
		packet.setRoomId(roomId);
		packet.setTimestamp(System.currentTimeMillis() / 1000);
		packet.setMessageIndex(messageIndex);

		messageDAO.saveMessage(roomId, packet);	//Redis 저장
		messageDAO.sendMessage(roomId, packet);	//메세지 전송 (publish)

		//응답
		result = new JSONObject();
		result.put("result", true);
		result.put("type", Protocol.createRoom.name());
		JSONObject body = new JSONObject();
		body.put("roomId", roomId);
		result.put("response", body);

	}catch(Exception e){
		e.printStackTrace();

		//응답
		result = new JSONObject();
		try {
			result.put("result", false);
			result.put("type", Protocol.createRoom.name());
		} catch (JSONException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
	}finally{
		ch.writeAndFlush(result.toString());
	}
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
@Bean
ChannelTopic sampleTopic() {
	return new ChannelTopic( "CHANGIM" );
}
 

import org.springframework.data.redis.listener.ChannelTopic; //导入依赖的package包/类
@Bean
ChannelTopic createTopic()
{
    return new ChannelTopic("pubsub:channel");
}