001 /*
002 *
003 * Licensed to the Apache Software Foundation (ASF) under one
004 * or more contributor license agreements. See the NOTICE file
005 * distributed with this work for additional information
006 * regarding copyright ownership. The ASF licenses this file
007 * to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance
009 * with the License. You may obtain a copy of the License at
010 *
011 * http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing,
014 * software distributed under the License is distributed on an
015 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
016 * KIND, either express or implied. See the License for the
017 * specific language governing permissions and limitations
018 * under the License.
019 *
020 */
021 package org.apache.qpid.jndi;
022
023 import java.io.BufferedInputStream;
024 import java.io.FileInputStream;
025 import java.io.IOException;
026 import java.net.URISyntaxException;
027 import java.util.Hashtable;
028 import java.util.Iterator;
029 import java.util.Map;
030 import java.util.Properties;
031 import java.util.concurrent.ConcurrentHashMap;
032
033 import javax.jms.ConnectionFactory;
034 import javax.jms.Destination;
035 import javax.jms.Queue;
036 import javax.jms.Topic;
037 import javax.naming.Context;
038 import javax.naming.NamingException;
039 import javax.naming.spi.InitialContextFactory;
040
041 import org.apache.qpid.client.AMQConnectionFactory;
042 import org.apache.qpid.client.AMQDestination;
043 import org.apache.qpid.client.AMQHeadersExchange;
044 import org.apache.qpid.client.AMQQueue;
045 import org.apache.qpid.client.AMQTopic;
046 import org.apache.qpid.exchange.ExchangeDefaults;
047 import org.apache.qpid.framing.AMQShortString;
048 import org.apache.qpid.url.AMQBindingURL;
049 import org.apache.qpid.url.BindingURL;
050 import org.apache.qpid.url.URLSyntaxException;
051 import org.slf4j.Logger;
052 import org.slf4j.LoggerFactory;
053
054 public class PropertiesFileInitialContextFactory implements InitialContextFactory
055 {
056 protected final Logger _logger = LoggerFactory.getLogger(PropertiesFileInitialContextFactory.class);
057
058 private String CONNECTION_FACTORY_PREFIX = "connectionfactory.";
059 private String DESTINATION_PREFIX = "destination.";
060 private String QUEUE_PREFIX = "queue.";
061 private String TOPIC_PREFIX = "topic.";
062
063 public Context getInitialContext(Hashtable environment) throws NamingException
064 {
065 Map data = new ConcurrentHashMap();
066
067 try
068 {
069
070 String file = null;
071 if (environment.containsKey(Context.PROVIDER_URL))
072 {
073 file = (String) environment.get(Context.PROVIDER_URL);
074 }
075 else
076 {
077 file = System.getProperty(Context.PROVIDER_URL);
078 }
079
080 if (file != null)
081 {
082 _logger.info("Loading Properties from:" + file);
083 // Load the properties specified
084 Properties p = new Properties();
085
086 p.load(new BufferedInputStream(new FileInputStream(file)));
087
088 environment.putAll(p);
089 System.getProperties().putAll(p);
090 _logger.info("Loaded Context Properties:" + environment.toString());
091 }
092 else
093 {
094 _logger.info("No Provider URL specified.");
095 }
096 }
097 catch (IOException ioe)
098 {
099 _logger.warn("Unable to load property file specified in Provider_URL:" + environment.get(Context.PROVIDER_URL));
100 }
101
102 createConnectionFactories(data, environment);
103
104 createDestinations(data, environment);
105
106 createQueues(data, environment);
107
108 createTopics(data, environment);
109
110 return createContext(data, environment);
111 }
112
113 // Implementation methods
114 // -------------------------------------------------------------------------
115 protected ReadOnlyContext createContext(Map data, Hashtable environment)
116 {
117 return new ReadOnlyContext(environment, data);
118 }
119
120 protected void createConnectionFactories(Map data, Hashtable environment)
121 {
122 for (Iterator iter = environment.entrySet().iterator(); iter.hasNext();)
123 {
124 Map.Entry entry = (Map.Entry) iter.next();
125 String key = entry.getKey().toString();
126 if (key.startsWith(CONNECTION_FACTORY_PREFIX))
127 {
128 String jndiName = key.substring(CONNECTION_FACTORY_PREFIX.length());
129 ConnectionFactory cf = createFactory(entry.getValue().toString());
130 if (cf != null)
131 {
132 data.put(jndiName, cf);
133 }
134 }
135 }
136 }
137
138 protected void createDestinations(Map data, Hashtable environment)
139 {
140 for (Iterator iter = environment.entrySet().iterator(); iter.hasNext();)
141 {
142 Map.Entry entry = (Map.Entry) iter.next();
143 String key = entry.getKey().toString();
144 if (key.startsWith(DESTINATION_PREFIX))
145 {
146 String jndiName = key.substring(DESTINATION_PREFIX.length());
147 Destination dest = createDestination(entry.getValue().toString());
148 if (dest != null)
149 {
150 data.put(jndiName, dest);
151 }
152 }
153 }
154 }
155
156 protected void createQueues(Map data, Hashtable environment)
157 {
158 for (Iterator iter = environment.entrySet().iterator(); iter.hasNext();)
159 {
160 Map.Entry entry = (Map.Entry) iter.next();
161 String key = entry.getKey().toString();
162 if (key.startsWith(QUEUE_PREFIX))
163 {
164 String jndiName = key.substring(QUEUE_PREFIX.length());
165 Queue q = createQueue(entry.getValue().toString());
166 if (q != null)
167 {
168 data.put(jndiName, q);
169 }
170 }
171 }
172 }
173
174 protected void createTopics(Map data, Hashtable environment)
175 {
176 for (Iterator iter = environment.entrySet().iterator(); iter.hasNext();)
177 {
178 Map.Entry entry = (Map.Entry) iter.next();
179 String key = entry.getKey().toString();
180 if (key.startsWith(TOPIC_PREFIX))
181 {
182 String jndiName = key.substring(TOPIC_PREFIX.length());
183 Topic t = createTopic(entry.getValue().toString());
184 if (t != null)
185 {
186 if (_logger.isDebugEnabled())
187 {
188 StringBuffer b = new StringBuffer();
189 b.append("Creating the topic: " + jndiName + " with the following binding keys ");
190 for (AMQShortString binding:((AMQTopic)t).getBindingKeys())
191 {
192 b.append(binding.toString()).append(",");
193 }
194
195 _logger.debug(b.toString());
196 }
197 data.put(jndiName, t);
198 }
199 }
200 }
201 }
202
203 /**
204 * Factory method to create new Connection Factory instances
205 */
206 protected ConnectionFactory createFactory(String url)
207 {
208 try
209 {
210 return new AMQConnectionFactory(url);
211 }
212 catch (URLSyntaxException urlse)
213 {
214 _logger.warn("Unable to createFactories:" + urlse);
215 }
216
217 return null;
218 }
219
220 /**
221 * Factory method to create new Destination instances from an AMQP BindingURL
222 */
223 protected Destination createDestination(String bindingURL)
224 {
225 AMQBindingURL binding;
226 try
227 {
228 binding = new AMQBindingURL(bindingURL);
229 }
230 catch (URISyntaxException urlse)
231 {
232 _logger.warn("Unable to create destination:" + urlse, urlse);
233
234 return null;
235 }
236
237 try
238 {
239 return AMQDestination.createDestination(binding);
240 }
241 catch (IllegalArgumentException iaw)
242 {
243 _logger.warn("Binding: '" + binding + "' not supported");
244
245 return null;
246 }
247 }
248
249 /**
250 * Factory method to create new Queue instances
251 */
252 protected Queue createQueue(Object value)
253 {
254 if (value instanceof AMQShortString)
255 {
256 return new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, (AMQShortString) value);
257 }
258 else if (value instanceof String)
259 {
260 return new AMQQueue(ExchangeDefaults.DIRECT_EXCHANGE_NAME, new AMQShortString((String) value));
261 }
262 else if (value instanceof BindingURL)
263 {
264 return new AMQQueue((BindingURL) value);
265 }
266
267 return null;
268 }
269
270 /**
271 * Factory method to create new Topic instances
272 */
273 protected Topic createTopic(Object value)
274 {
275 if (value instanceof AMQShortString)
276 {
277 return new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME, (AMQShortString) value);
278 }
279 else if (value instanceof String)
280 {
281 String[] keys = ((String)value).split(",");
282 AMQShortString[] bindings = new AMQShortString[keys.length];
283 int i = 0;
284 for (String key:keys)
285 {
286 bindings[i] = new AMQShortString(key);
287 i++;
288 }
289 // The Destination has a dual nature. If this was used for a producer the key is used
290 // for the routing key. If it was used for the consumer it becomes the bindingKey
291 return new AMQTopic(ExchangeDefaults.TOPIC_EXCHANGE_NAME,bindings[0],null,bindings);
292 }
293 else if (value instanceof BindingURL)
294 {
295 return new AMQTopic((BindingURL) value);
296 }
297
298 return null;
299 }
300
301 /**
302 * Factory method to create new HeaderExcahnge instances
303 */
304 protected Destination createHeaderExchange(Object value)
305 {
306 if (value instanceof String)
307 {
308 return new AMQHeadersExchange((String) value);
309 }
310 else if (value instanceof BindingURL)
311 {
312 return new AMQHeadersExchange((BindingURL) value);
313 }
314
315 return null;
316 }
317
318 // Properties
319 // -------------------------------------------------------------------------
320 public String getConnectionPrefix()
321 {
322 return CONNECTION_FACTORY_PREFIX;
323 }
324
325 public void setConnectionPrefix(String connectionPrefix)
326 {
327 this.CONNECTION_FACTORY_PREFIX = connectionPrefix;
328 }
329
330 public String getDestinationPrefix()
331 {
332 return DESTINATION_PREFIX;
333 }
334
335 public void setDestinationPrefix(String destinationPrefix)
336 {
337 this.DESTINATION_PREFIX = destinationPrefix;
338 }
339
340 public String getQueuePrefix()
341 {
342 return QUEUE_PREFIX;
343 }
344
345 public void setQueuePrefix(String queuePrefix)
346 {
347 this.QUEUE_PREFIX = queuePrefix;
348 }
349
350 public String getTopicPrefix()
351 {
352 return TOPIC_PREFIX;
353 }
354
355 public void setTopicPrefix(String topicPrefix)
356 {
357 this.TOPIC_PREFIX = topicPrefix;
358 }
359 }
|