Webinar End State

The state of the project at the end of the webinar,
use git log -p <this commit> to see what changed.

intermediate/monitoring/src/main/java/org/springintegration/PayloadAwareTimingInterceptor.java

@@ -0,0 +1,164 @@
+/*
+ * Copyright 2002-2012 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springintegration;
+
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
+import org.springframework.jmx.export.annotation.ManagedOperation;
+import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.util.StopWatch;
+
+/**
+ * A sample channel interceptor that illustrates a technique to capture elapsed times
+ * based on message payload types.
+ *
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+@ManagedResource
+public class PayloadAwareTimingInterceptor extends ChannelInterceptorAdapter {
+
+	private ThreadLocal<StopWatchHolder> stopWatchHolder = new ThreadLocal<PayloadAwareTimingInterceptor.StopWatchHolder>();
+
+	private Map<Class<?>, Stats> statsMap = new ConcurrentHashMap<Class<?>, PayloadAwareTimingInterceptor.Stats>();
+
+	/**
+	 *
+	 * @param classes An array of types for which statistics will be captured; if
+	 * not supplied {@link Object} will be added as a catch-all.
+	 */
+	public PayloadAwareTimingInterceptor(Class<?>[] classes) {
+		for (Class<?> clazz : classes) {
+			this.statsMap.put(clazz, new Stats());
+		}
+		if (!this.statsMap.containsKey(Object.class)) {
+			this.statsMap.put(Object.class, new Stats());
+		}
+	}
+
+	@Override
+	public Message<?> preSend(Message<?> message, MessageChannel channel) {
+		StopWatch stopWatch = new StopWatch();
+		stopWatch.start();
+		this.stopWatchHolder.set(new StopWatchHolder(stopWatch, message.getPayload().getClass()));
+		return super.preSend(message, channel);
+	}
+
+	@Override
+	public void postSend(Message<?> message, MessageChannel channel, boolean sent) {
+		StopWatchHolder holder = this.stopWatchHolder.get();
+		if (holder != null) {
+			holder.getStopWatch().stop();
+		}
+		Stats stats = this.statsMap.get(holder.getType());
+		if (stats == null) {
+			stats = this.statsMap.get(Object.class);
+		}
+		stats.add(holder.getStopWatch().getLastTaskTimeMillis());
+	}
+
+	@ManagedOperation
+	public String[] getSummary() {
+		String[] data = new String[this.statsMap.size()];
+		int i = 0;
+		for (Entry<Class<?>, Stats> entry : this.statsMap.entrySet()) {
+			data[i++] = entry.getKey().getName() + " " + entry.getValue().toString();
+		}
+		return data;
+	}
+
+	@ManagedOperation
+	public long getCount(String className) throws Exception {
+		return this.statsMap.get(Class.forName(className)).getCount();
+	}
+
+	@ManagedOperation
+	public long getLastTime(String className) throws Exception {
+		return this.statsMap.get(Class.forName(className)).getLastTime();
+	}
+
+	@ManagedOperation
+	public float getAverage(String className) throws Exception {
+		return this.statsMap.get(Class.forName(className)).getAverage();
+	}
+
+	private class StopWatchHolder {
+
+		private final StopWatch stopWatch;
+
+		private final Class<?> type;
+
+		/**
+		 * @param stopWatch
+		 * @param type
+		 */
+		public StopWatchHolder(StopWatch stopWatch, Class<?> type) {
+			this.stopWatch = stopWatch;
+			this.type = type;
+		}
+
+		public StopWatch getStopWatch() {
+			return stopWatch;
+		}
+
+		public Class<?> getType() {
+			return type;
+		}
+	}
+
+	private class Stats {
+
+		private long count;
+
+		private long totalTime;
+
+		private float average;
+
+		private long lastTime;
+
+		public long getCount() {
+			return count;
+		}
+
+		public long getLastTime() {
+			return lastTime;
+		}
+
+		public float getAverage() {
+			return average;
+		}
+
+		public synchronized void add(long time) {
+			this.count++;
+			this.lastTime = time;
+			this.totalTime += time;
+			this.average = (float) this.totalTime / (float) this.count;
+		}
+
+		@Override
+		public String toString() {
+			return "Stats [count=" + count + ", average=" + average + ", lastTime=" + lastTime + "]";
+		}
+
+
+	}
+}

intermediate/monitoring/src/main/java/org/springintegration/service/impl/DefaultTwitterService.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2002-2012 the original author or authors
+ *
+ *	 Licensed under the Apache License, Version 2.0 (the "License");
+ *	 you may not use this file except in compliance with the License.
+ *	 You may obtain a copy of the License at
+ *
+ *		 http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *	 Unless required by applicable law or agreed to in writing, software
+ *	 distributed under the License is distributed on an "AS IS" BASIS,
+ *	 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *	 See the License for the specific language governing permissions and
+ *	 limitations under the License.
+ */
+package org.springintegration.service.impl;
+
+import java.util.Collection;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.model.TwitterMessage;
+import org.springframework.integration.service.TwitterService;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.export.annotation.ManagedResource;
+import org.springframework.jmx.support.MetricType;
+import org.springframework.stereotype.Service;
+
+/**
+ * Implementation of the TwitterService interface.
+ * In the org.springintegration package because SI excludes all org.springframework.integration.*
+ * classes from MBean export.
+ */
+@Service
+@ManagedResource
+public class DefaultTwitterService implements TwitterService {
+
+	/** Holds a collection of polled Twitter messages */
+	private Map<Long, TwitterMessage> twitterMessages;
+
+	@Autowired
+	private MessageChannel controlBusChannel;
+
+	private long totalTweets;
+
+	@Autowired(required=false)
+	@Qualifier("dummyTwitter")
+	private SourcePollingChannelAdapter dummyTwitter;
+
+	@Autowired(required=false)
+	private IntegrationMBeanExporter exporter;
+	/**
+	 * Constructor that initializes the 'twitterMessages' Map as a simple LRU
+	 * cache. @See http://blogs.oracle.com/swinger/entry/collections_trick_i_lru_cache
+	 */
+	public DefaultTwitterService() {
+
+		twitterMessages = new LinkedHashMap<Long, TwitterMessage>( 10, 0.75f, true ) {
+
+			private static final long serialVersionUID = 1L;
+
+			@Override
+			protected boolean removeEldestEntry( java.util.Map.Entry<Long, TwitterMessage> entry ) {
+				return size() > 10;
+			}
+
+		};
+
+	}
+
+
+	/**
+	 * @return the totalTweets
+	 */
+	@ManagedMetric(metricType=MetricType.COUNTER)
+	public long getTotalTweets() {
+		return totalTweets;
+	}
+
+
+	/** {@inheritDoc} */
+	@Override
+	public Collection<TwitterMessage> getTwitterMessages() {
+		return twitterMessages.values();
+	}
+
+	/** {@inheritDoc} */
+	@Override
+	public void startTwitterAdapter() {
+
+		Message<String> operation = MessageBuilder.withPayload("@twitter.start()").build();
+
+		this.controlBusChannel.send(operation);
+
+		if (this.dummyTwitter != null) {
+			this.controlBusChannel.send(MessageBuilder.withPayload("@dummyTwitter.start()").build());
+		}
+	}
+
+	/** {@inheritDoc} */
+	@Override
+	public void stopTwitterAdapter() {
+
+		Message<String> operation = MessageBuilder.withPayload("@twitter.stop()").build();
+
+		this.controlBusChannel.send(operation);
+
+		if (this.dummyTwitter != null) {
+			this.controlBusChannel.send(MessageBuilder.withPayload("@dummyTwitter.stop()").build());
+		}
+	}
+
+
+	@Override
+	public void shutdown() {
+		Message<String> operation = MessageBuilder.withPayload("@integrationMBeanExporter.stopActiveComponents(false, 20000)").build();
+
+		if (this.exporter != null) {
+			this.controlBusChannel.send(operation);
+		}
+	}
+
+
+	/**
+	 * Called by Spring Integration to populate a simple LRU cache.
+	 *
+	 * @param tweet - The Spring Integration tweet object.
+	 * @throws InterruptedException
+	 */
+	public void addTwitterMessages(Tweet tweet) throws Exception {
+		if ("SomeUser".equals(tweet.getFromUser())) {
+			Thread.sleep(2000);
+		}
+		this.totalTweets++;
+		this.twitterMessages.put(tweet.getCreatedAt().getTime(), new TwitterMessage(tweet.getCreatedAt(),
+				tweet.getText(),
+				tweet.getFromUser(),
+				tweet.getProfileImageUrl()));
+	}
+
+}