001    /**
002     * Licensed to the Apache Software Foundation (ASF) under one or more
003     * contributor license agreements.  See the NOTICE file distributed with
004     * this work for additional information regarding copyright ownership.
005     * The ASF licenses this file to You under the Apache License, Version 2.0
006     * (the "License"); you may not use this file except in compliance with
007     * the License.  You may obtain a copy of the License at
008     *
009     *      http://www.apache.org/licenses/LICENSE-2.0
010     *
011     * Unless required by applicable law or agreed to in writing, software
012     * distributed under the License is distributed on an "AS IS" BASIS,
013     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014     * See the License for the specific language governing permissions and
015     * limitations under the License.
016     */
017    package org.apache.activemq.broker.jmx;
018    
019    import java.io.IOException;
020    import java.util.ArrayList;
021    import java.util.Date;
022    import java.util.HashMap;
023    import java.util.List;
024    import java.util.Map;
025    import java.util.Set;
026    import javax.jms.DeliveryMode;
027    import javax.jms.JMSException;
028    import javax.management.openmbean.ArrayType;
029    import javax.management.openmbean.CompositeData;
030    import javax.management.openmbean.CompositeDataSupport;
031    import javax.management.openmbean.CompositeType;
032    import javax.management.openmbean.OpenDataException;
033    import javax.management.openmbean.OpenType;
034    import javax.management.openmbean.SimpleType;
035    import javax.management.openmbean.TabularDataSupport;
036    import javax.management.openmbean.TabularType;
037    
038    import org.apache.activemq.broker.region.policy.SlowConsumerEntry;
039    import org.apache.activemq.broker.scheduler.Job;
040    import org.apache.activemq.command.ActiveMQBlobMessage;
041    import org.apache.activemq.command.ActiveMQBytesMessage;
042    import org.apache.activemq.command.ActiveMQMapMessage;
043    import org.apache.activemq.command.ActiveMQMessage;
044    import org.apache.activemq.command.ActiveMQObjectMessage;
045    import org.apache.activemq.command.ActiveMQStreamMessage;
046    import org.apache.activemq.command.ActiveMQTextMessage;
047    
048    public final class OpenTypeSupport {
049    
050        interface OpenTypeFactory {
051            CompositeType getCompositeType() throws OpenDataException;
052    
053            Map<String, Object> getFields(Object o) throws OpenDataException;
054        }
055    
056        private static final Map<Class, AbstractOpenTypeFactory> OPEN_TYPE_FACTORIES = new HashMap<Class, AbstractOpenTypeFactory>();
057    
058        abstract static class AbstractOpenTypeFactory implements OpenTypeFactory {
059    
060            private CompositeType compositeType;
061            private final List<String> itemNamesList = new ArrayList<String>();
062            private final List<String> itemDescriptionsList = new ArrayList<String>();
063            private final List<OpenType> itemTypesList = new ArrayList<OpenType>();
064    
065            public CompositeType getCompositeType() throws OpenDataException {
066                if (compositeType == null) {
067                    init();
068                    compositeType = createCompositeType();
069                }
070                return compositeType;
071            }
072    
073            protected void init() throws OpenDataException {
074            }
075    
076            protected CompositeType createCompositeType() throws OpenDataException {
077                String[] itemNames = itemNamesList.toArray(new String[itemNamesList.size()]);
078                String[] itemDescriptions = itemDescriptionsList.toArray(new String[itemDescriptionsList.size()]);
079                OpenType[] itemTypes = itemTypesList.toArray(new OpenType[itemTypesList.size()]);
080                return new CompositeType(getTypeName(), getDescription(), itemNames, itemDescriptions, itemTypes);
081            }
082    
083            protected abstract String getTypeName();
084    
085            protected void addItem(String name, String description, OpenType type) {
086                itemNamesList.add(name);
087                itemDescriptionsList.add(description);
088                itemTypesList.add(type);
089            }
090    
091            protected String getDescription() {
092                return getTypeName();
093            }
094    
095            public Map<String, Object> getFields(Object o) throws OpenDataException {
096                Map<String, Object> rc = new HashMap<String, Object>();
097                return rc;
098            }
099        }
100    
101        static class MessageOpenTypeFactory extends AbstractOpenTypeFactory {
102            protected TabularType stringPropertyTabularType;
103            protected TabularType booleanPropertyTabularType;
104            protected TabularType bytePropertyTabularType;
105            protected TabularType shortPropertyTabularType;
106            protected TabularType intPropertyTabularType;
107            protected TabularType longPropertyTabularType;
108            protected TabularType floatPropertyTabularType;
109            protected TabularType doublePropertyTabularType;
110    
111            @Override
112            protected String getTypeName() {
113                return ActiveMQMessage.class.getName();
114            }
115    
116            @Override
117            protected void init() throws OpenDataException {
118                super.init();
119                addItem("JMSCorrelationID", "JMSCorrelationID", SimpleType.STRING);
120                addItem("JMSDestination", "JMSDestination", SimpleType.STRING);
121                addItem("JMSMessageID", "JMSMessageID", SimpleType.STRING);
122                addItem("JMSReplyTo", "JMSReplyTo", SimpleType.STRING);
123                addItem("JMSType", "JMSType", SimpleType.STRING);
124                addItem("JMSDeliveryMode", "JMSDeliveryMode", SimpleType.STRING);
125                addItem("JMSExpiration", "JMSExpiration", SimpleType.LONG);
126                addItem("JMSPriority", "JMSPriority", SimpleType.INTEGER);
127                addItem("JMSRedelivered", "JMSRedelivered", SimpleType.BOOLEAN);
128                addItem("JMSTimestamp", "JMSTimestamp", SimpleType.DATE);
129                addItem(CompositeDataConstants.JMSXGROUP_ID, "Message Group ID", SimpleType.STRING);
130                addItem(CompositeDataConstants.JMSXGROUP_SEQ, "Message Group Sequence Number", SimpleType.INTEGER);
131                addItem(CompositeDataConstants.ORIGINAL_DESTINATION, "Original Destination Before Senting To DLQ", SimpleType.STRING);
132                addItem(CompositeDataConstants.PROPERTIES, "User Properties Text", SimpleType.STRING);
133    
134                // now lets expose the type safe properties
135                stringPropertyTabularType = createTabularType(String.class, SimpleType.STRING);
136                booleanPropertyTabularType = createTabularType(Boolean.class, SimpleType.BOOLEAN);
137                bytePropertyTabularType = createTabularType(Byte.class, SimpleType.BYTE);
138                shortPropertyTabularType = createTabularType(Short.class, SimpleType.SHORT);
139                intPropertyTabularType = createTabularType(Integer.class, SimpleType.INTEGER);
140                longPropertyTabularType = createTabularType(Long.class, SimpleType.LONG);
141                floatPropertyTabularType = createTabularType(Float.class, SimpleType.FLOAT);
142                doublePropertyTabularType = createTabularType(Double.class, SimpleType.DOUBLE);
143    
144                addItem(CompositeDataConstants.STRING_PROPERTIES, "User String Properties", stringPropertyTabularType);
145                addItem(CompositeDataConstants.BOOLEAN_PROPERTIES, "User Boolean Properties", booleanPropertyTabularType);
146                addItem(CompositeDataConstants.BYTE_PROPERTIES, "User Byte Properties", bytePropertyTabularType);
147                addItem(CompositeDataConstants.SHORT_PROPERTIES, "User Short Properties", shortPropertyTabularType);
148                addItem(CompositeDataConstants.INT_PROPERTIES, "User Integer Properties", intPropertyTabularType);
149                addItem(CompositeDataConstants.LONG_PROPERTIES, "User Long Properties", longPropertyTabularType);
150                addItem(CompositeDataConstants.FLOAT_PROPERTIES, "User Float Properties", floatPropertyTabularType);
151                addItem(CompositeDataConstants.DOUBLE_PROPERTIES, "User Double Properties", doublePropertyTabularType);
152            }
153    
154            @Override
155            public Map<String, Object> getFields(Object o) throws OpenDataException {
156                ActiveMQMessage m = (ActiveMQMessage)o;
157                Map<String, Object> rc = super.getFields(o);
158                rc.put("JMSCorrelationID", m.getJMSCorrelationID());
159                rc.put("JMSDestination", "" + m.getJMSDestination());
160                rc.put("JMSMessageID", m.getJMSMessageID());
161                rc.put("JMSReplyTo",toString(m.getJMSReplyTo()));
162                rc.put("JMSType", m.getJMSType());
163                rc.put("JMSDeliveryMode", m.getJMSDeliveryMode() == DeliveryMode.PERSISTENT ? "PERSISTENT" : "NON-PERSISTENT");
164                rc.put("JMSExpiration", Long.valueOf(m.getJMSExpiration()));
165                rc.put("JMSPriority", Integer.valueOf(m.getJMSPriority()));
166                rc.put("JMSRedelivered", Boolean.valueOf(m.getJMSRedelivered()));
167                rc.put("JMSTimestamp", new Date(m.getJMSTimestamp()));
168                rc.put(CompositeDataConstants.JMSXGROUP_ID, m.getGroupID());
169                rc.put(CompositeDataConstants.JMSXGROUP_SEQ, m.getGroupSequence());
170                rc.put(CompositeDataConstants.ORIGINAL_DESTINATION, toString(m.getOriginalDestination()));
171                try {
172                    rc.put(CompositeDataConstants.PROPERTIES, "" + m.getProperties());
173                } catch (IOException e) {
174                    rc.put(CompositeDataConstants.PROPERTIES, "");
175                }
176    
177                try {
178                    rc.put(CompositeDataConstants.STRING_PROPERTIES, createTabularData(m, stringPropertyTabularType, String.class));
179                } catch (IOException e) {
180                    rc.put(CompositeDataConstants.STRING_PROPERTIES, new TabularDataSupport(stringPropertyTabularType));
181                }
182                try {
183                    rc.put(CompositeDataConstants.BOOLEAN_PROPERTIES, createTabularData(m, booleanPropertyTabularType, Boolean.class));
184                } catch (IOException e) {
185                    rc.put(CompositeDataConstants.BOOLEAN_PROPERTIES, new TabularDataSupport(booleanPropertyTabularType));
186                }
187                try {
188                    rc.put(CompositeDataConstants.BYTE_PROPERTIES, createTabularData(m, bytePropertyTabularType, Byte.class));
189                } catch (IOException e) {
190                    rc.put(CompositeDataConstants.BYTE_PROPERTIES, new TabularDataSupport(bytePropertyTabularType));
191                }
192                try {
193                    rc.put(CompositeDataConstants.SHORT_PROPERTIES, createTabularData(m, shortPropertyTabularType, Short.class));
194                } catch (IOException e) {
195                    rc.put(CompositeDataConstants.SHORT_PROPERTIES, new TabularDataSupport(shortPropertyTabularType));
196                }
197                try {
198                    rc.put(CompositeDataConstants.INT_PROPERTIES, createTabularData(m, intPropertyTabularType, Integer.class));
199                } catch (IOException e) {
200                    rc.put(CompositeDataConstants.INT_PROPERTIES, new TabularDataSupport(intPropertyTabularType));
201                }
202                try {
203                    rc.put(CompositeDataConstants.LONG_PROPERTIES, createTabularData(m, longPropertyTabularType, Long.class));
204                } catch (IOException e) {
205                    rc.put(CompositeDataConstants.LONG_PROPERTIES, new TabularDataSupport(longPropertyTabularType));
206                }
207                try {
208                    rc.put(CompositeDataConstants.FLOAT_PROPERTIES, createTabularData(m, floatPropertyTabularType, Float.class));
209                } catch (IOException e) {
210                    rc.put(CompositeDataConstants.FLOAT_PROPERTIES, new TabularDataSupport(floatPropertyTabularType));
211                }
212                try {
213                    rc.put(CompositeDataConstants.DOUBLE_PROPERTIES, createTabularData(m, doublePropertyTabularType, Double.class));
214                } catch (IOException e) {
215                    rc.put(CompositeDataConstants.DOUBLE_PROPERTIES, new TabularDataSupport(doublePropertyTabularType));
216                }
217                return rc;
218            }
219    
220            protected String toString(Object value) {
221                if (value == null) {
222                    return null;
223                }
224                return value.toString();
225            }
226    
227    
228            protected <T> TabularType createTabularType(Class<T> type, OpenType openType) throws OpenDataException {
229                String typeName = "java.util.Map<java.lang.String, " + type.getName() + ">";
230                String[] keyValue = new String[]{"key", "value"};
231                OpenType[] openTypes = new OpenType[]{SimpleType.STRING, openType};
232                CompositeType rowType = new CompositeType(typeName, typeName, keyValue, keyValue, openTypes);
233                return new TabularType(typeName, typeName, rowType, new String[]{"key"});
234            }
235    
236            protected TabularDataSupport createTabularData(ActiveMQMessage m, TabularType type, Class valueType) throws IOException, OpenDataException {
237                TabularDataSupport answer = new TabularDataSupport(type);
238                Set<Map.Entry<String,Object>> entries = m.getProperties().entrySet();
239                for (Map.Entry<String, Object> entry : entries) {
240                    Object value = entry.getValue();
241                    if (valueType.isInstance(value)) {
242                        CompositeDataSupport compositeData = createTabularRowValue(type, entry.getKey(), value);
243                        answer.put(compositeData);
244                    }
245                }
246                return answer;
247            }
248    
249            protected CompositeDataSupport createTabularRowValue(TabularType type, String key, Object value) throws OpenDataException {
250                Map<String,Object> fields = new HashMap<String, Object>();
251                fields.put("key", key);
252                fields.put("value", value);
253                return new CompositeDataSupport(type.getRowType(), fields);
254            }
255        }
256    
257        static class ByteMessageOpenTypeFactory extends MessageOpenTypeFactory {
258    
259    
260            @Override
261            protected String getTypeName() {
262                return ActiveMQBytesMessage.class.getName();
263            }
264    
265            @Override
266            protected void init() throws OpenDataException {
267                super.init();
268                addItem(CompositeDataConstants.BODY_LENGTH, "Body length", SimpleType.LONG);
269                addItem(CompositeDataConstants.BODY_PREVIEW, "Body preview", new ArrayType(1, SimpleType.BYTE));
270            }
271    
272            @Override
273            public Map<String, Object> getFields(Object o) throws OpenDataException {
274                ActiveMQBytesMessage m = (ActiveMQBytesMessage)o;
275                m.setReadOnlyBody(true);
276                Map<String, Object> rc = super.getFields(o);
277                long length = 0;
278                try {
279                    length = m.getBodyLength();
280                    rc.put(CompositeDataConstants.BODY_LENGTH, Long.valueOf(length));
281                } catch (JMSException e) {
282                    rc.put(CompositeDataConstants.BODY_LENGTH, Long.valueOf(0));
283                }
284                try {
285                    byte preview[] = new byte[(int)Math.min(length, 255)];
286                    m.readBytes(preview);
287    
288                    // This is whack! Java 1.5 JMX spec does not support primitive
289                    // arrays!
290                    // In 1.6 it seems it is supported.. but until then...
291                    Byte data[] = new Byte[preview.length];
292                    for (int i = 0; i < data.length; i++) {
293                        data[i] = new Byte(preview[i]);
294                    }
295    
296                    rc.put(CompositeDataConstants.BODY_PREVIEW, data);
297                } catch (JMSException e) {
298                    rc.put(CompositeDataConstants.BODY_PREVIEW, new Byte[] {});
299                }
300                return rc;
301            }
302    
303        }
304    
305        static class MapMessageOpenTypeFactory extends MessageOpenTypeFactory {
306    
307            @Override
308            protected String getTypeName() {
309                return ActiveMQMapMessage.class.getName();
310            }
311    
312            @Override
313            protected void init() throws OpenDataException {
314                super.init();
315                addItem(CompositeDataConstants.CONTENT_MAP, "Content map", SimpleType.STRING);
316            }
317    
318            @Override
319            public Map<String, Object> getFields(Object o) throws OpenDataException {
320                ActiveMQMapMessage m = (ActiveMQMapMessage)o;
321                Map<String, Object> rc = super.getFields(o);
322                try {
323                    rc.put(CompositeDataConstants.CONTENT_MAP, "" + m.getContentMap());
324                } catch (JMSException e) {
325                    rc.put(CompositeDataConstants.CONTENT_MAP, "");
326                }
327                return rc;
328            }
329        }
330    
331        static class ObjectMessageOpenTypeFactory extends MessageOpenTypeFactory {
332            @Override
333            protected String getTypeName() {
334                return ActiveMQObjectMessage.class.getName();
335            }
336    
337            @Override
338            protected void init() throws OpenDataException {
339                super.init();
340            }
341    
342            @Override
343            public Map<String, Object> getFields(Object o) throws OpenDataException {
344                Map<String, Object> rc = super.getFields(o);
345                return rc;
346            }
347        }
348    
349        static class StreamMessageOpenTypeFactory extends MessageOpenTypeFactory {
350            @Override
351            protected String getTypeName() {
352                return ActiveMQStreamMessage.class.getName();
353            }
354    
355            @Override
356            protected void init() throws OpenDataException {
357                super.init();
358            }
359    
360            @Override
361            public Map<String, Object> getFields(Object o) throws OpenDataException {
362                Map<String, Object> rc = super.getFields(o);
363                return rc;
364            }
365        }
366    
367        static class TextMessageOpenTypeFactory extends MessageOpenTypeFactory {
368    
369            @Override
370            protected String getTypeName() {
371                return ActiveMQTextMessage.class.getName();
372            }
373    
374            @Override
375            protected void init() throws OpenDataException {
376                super.init();
377                addItem(CompositeDataConstants.MESSAGE_TEXT, CompositeDataConstants.MESSAGE_TEXT, SimpleType.STRING);
378            }
379    
380            @Override
381            public Map<String, Object> getFields(Object o) throws OpenDataException {
382                ActiveMQTextMessage m = (ActiveMQTextMessage)o;
383                Map<String, Object> rc = super.getFields(o);
384                try {
385                    rc.put(CompositeDataConstants.MESSAGE_TEXT, "" + m.getText());
386                } catch (JMSException e) {
387                    rc.put(CompositeDataConstants.MESSAGE_TEXT, "");
388                }
389                return rc;
390            }
391        }
392        
393    
394        static class JobOpenTypeFactory extends AbstractOpenTypeFactory {
395    
396            @Override
397            protected String getTypeName() {
398                return Job.class.getName();
399            }
400    
401            @Override
402            protected void init() throws OpenDataException {
403                super.init();
404                addItem("jobId", "jobId", SimpleType.STRING);
405                addItem("cronEntry", "Cron entry", SimpleType.STRING);
406                addItem("start", "start time", SimpleType.STRING);
407                addItem("delay", "initial delay", SimpleType.LONG);
408                addItem("next", "next time", SimpleType.STRING);
409                addItem("period", "period between jobs", SimpleType.LONG);
410                addItem("repeat", "number of times to repeat", SimpleType.INTEGER);
411            }
412    
413            @Override
414            public Map<String, Object> getFields(Object o) throws OpenDataException {
415                Job job = (Job) o;
416                Map<String, Object> rc = super.getFields(o);
417                rc.put("jobId", job.getJobId());
418                rc.put("cronEntry", "" + job.getCronEntry());
419                rc.put("start", job.getStartTime());
420                rc.put("delay", job.getDelay());
421                rc.put("next", job.getNextExecutionTime());
422                rc.put("period", job.getPeriod());
423                rc.put("repeat", job.getRepeat());
424                return rc;
425            }
426        }
427    
428        static class ActiveMQBlobMessageOpenTypeFactory extends MessageOpenTypeFactory {
429    
430            @Override
431            protected String getTypeName() {
432                return ActiveMQBlobMessage.class.getName();
433            }
434    
435            @Override
436            protected void init() throws OpenDataException {
437                super.init();
438                addItem(CompositeDataConstants.MESSAGE_URL, "Body Url", SimpleType.STRING);
439            }
440    
441            @Override
442            public Map<String, Object> getFields(Object o) throws OpenDataException {
443                ActiveMQBlobMessage m = (ActiveMQBlobMessage)o;
444                Map<String, Object> rc = super.getFields(o);
445                try {
446                    rc.put(CompositeDataConstants.MESSAGE_URL, "" + m.getURL().toString());
447                } catch (JMSException e) {
448                    rc.put(CompositeDataConstants.MESSAGE_URL, "");
449                }
450                return rc;
451            }
452        }
453    
454        static class SlowConsumerEntryOpenTypeFactory extends AbstractOpenTypeFactory {
455           @Override
456            protected String getTypeName() {
457                return SlowConsumerEntry.class.getName();
458            }
459    
460            @Override
461            protected void init() throws OpenDataException {
462                super.init();
463                addItem("subscription", "the subscription view", SimpleType.OBJECTNAME);
464                addItem("slowCount", "number of times deemed slow", SimpleType.INTEGER);
465                addItem("markCount", "number of periods remaining slow", SimpleType.INTEGER);
466            }
467    
468            @Override
469            public Map<String, Object> getFields(Object o) throws OpenDataException {
470                SlowConsumerEntry entry = (SlowConsumerEntry) o;
471                Map<String, Object> rc = super.getFields(o);
472                rc.put("subscription", entry.getSubscription());
473                rc.put("slowCount", Integer.valueOf(entry.getSlowCount()));
474                rc.put("markCount", Integer.valueOf(entry.getMarkCount()));
475                return rc;
476            }
477        }
478    
479        static {
480            OPEN_TYPE_FACTORIES.put(ActiveMQMessage.class, new MessageOpenTypeFactory());
481            OPEN_TYPE_FACTORIES.put(ActiveMQBytesMessage.class, new ByteMessageOpenTypeFactory());
482            OPEN_TYPE_FACTORIES.put(ActiveMQMapMessage.class, new MapMessageOpenTypeFactory());
483            OPEN_TYPE_FACTORIES.put(ActiveMQObjectMessage.class, new ObjectMessageOpenTypeFactory());
484            OPEN_TYPE_FACTORIES.put(ActiveMQStreamMessage.class, new StreamMessageOpenTypeFactory());
485            OPEN_TYPE_FACTORIES.put(ActiveMQTextMessage.class, new TextMessageOpenTypeFactory());
486            OPEN_TYPE_FACTORIES.put(Job.class, new JobOpenTypeFactory());
487            OPEN_TYPE_FACTORIES.put(SlowConsumerEntry.class, new SlowConsumerEntryOpenTypeFactory());
488            OPEN_TYPE_FACTORIES.put(ActiveMQBlobMessage.class, new ActiveMQBlobMessageOpenTypeFactory());
489        }
490    
491        private OpenTypeSupport() {
492        }
493        
494        public static OpenTypeFactory getFactory(Class<?> clazz) throws OpenDataException {
495            return OPEN_TYPE_FACTORIES.get(clazz);
496        }
497    
498        public static CompositeData convert(Object message) throws OpenDataException {
499            OpenTypeFactory f = getFactory(message.getClass());
500            if (f == null) {
501                throw new OpenDataException("Cannot create a CompositeData for type: " + message.getClass().getName());
502            }
503            CompositeType ct = f.getCompositeType();
504            Map<String, Object> fields = f.getFields(message);
505            return new CompositeDataSupport(ct, fields);
506        }
507    
508    }