1 28 29 package com.caucho.jms.memory; 30 31 import com.caucho.jms.AbstractDestination; 32 import com.caucho.jms.session.SessionImpl; 33 import com.caucho.log.Log; 34 import com.caucho.util.L10N; 35 36 import javax.jms.JMSException ; 37 import javax.jms.Message ; 38 import javax.jms.MessageConsumer ; 39 import javax.jms.Topic ; 40 import javax.jms.TopicSubscriber ; 41 import java.util.ArrayList ; 42 import java.util.HashMap ; 43 import java.util.logging.Logger ; 44 45 48 public class MemoryTopic extends AbstractDestination implements Topic { 49 static final Logger log = Log.open(MemoryTopic.class); 50 static final L10N L = new L10N(MemoryTopic.class); 51 52 ArrayList <MemoryQueue> _subscribers = new ArrayList <MemoryQueue>(); 53 54 private HashMap <String ,MemoryQueue> _durableSubscribers = 55 new HashMap <String ,MemoryQueue>(); 56 57 private String _topicName; 58 59 public MemoryTopic() 60 { 61 } 62 63 66 public String getTopicName() 67 { 68 return _topicName; 69 } 70 71 74 public void setTopicName(String name) 75 { 76 _topicName = name; 77 } 78 79 public void send(Message message) 80 throws JMSException 81 { 82 for (int i = 0; i < _subscribers.size(); i++) { 83 MemoryQueue queue = _subscribers.get(i); 84 85 queue.send(message); 86 } 87 } 88 89 96 public MessageConsumer createConsumer(SessionImpl session, 97 String selector, 98 boolean noLocal) 99 throws JMSException 100 { 101 return new MemoryTopicConsumer(session, selector, this); 102 } 103 104 107 public TopicSubscriber createDurableSubscriber(SessionImpl session, 108 String selector, 109 boolean noLocal, 110 String name) 111 throws JMSException 112 { 113 return new MemoryTopicConsumer(session, selector, this, name); 114 } 115 116 119 public MemoryQueue createDurableSubscriber(String name) 120 throws JMSException 121 { 122 MemoryQueue queue = _durableSubscribers.get(name); 123 124 if (queue == null) { 125 queue = createSubscriberQueue(); 126 _durableSubscribers.put(name, queue); 127 } 128 129 return queue; 130 } 131 132 MemoryQueue createSubscriberQueue() 133 throws JMSException 134 { 135 MemoryQueue queue = new MemoryQueue(); 136 137 _subscribers.add(queue); 138 139 return queue; 140 } 141 142 public void removeSubscriber(MemoryQueue queue) 143 { 144 _subscribers.remove(queue); 145 } 146 147 150 public String toString() 151 { 152 return "Topic[" + _topicName + "]"; 153 } 154 } 155 156 | Popular Tags |