define spi, refine api with team

cat-core/src/main/java/com/dianping/cat/configuration/ComponentsConfigurator.java

@@ -6,14 +6,16 @@ import java.util.List;
 import com.dianping.cat.message.MessageProducer;
 import com.dianping.cat.message.broker.DefaultMessageBroker;
 import com.dianping.cat.message.broker.MessageBroker;
-import com.dianping.cat.message.handler.MessageDispatcher;
-import com.dianping.cat.message.handler.MessageHandler;
 import com.dianping.cat.message.internal.DefaultMessageProducer;
 import com.dianping.cat.message.io.InMemoryQueue;
 import com.dianping.cat.message.io.InMemoryReceiver;
 import com.dianping.cat.message.io.InMemorySender;
 import com.dianping.cat.message.io.MessageReceiver;
 import com.dianping.cat.message.io.MessageSender;
+import com.dianping.cat.message.spi.MessageCodec;
+import com.dianping.cat.message.spi.MessageHandler;
+import com.dianping.cat.message.spi.codec.PlainTextMessageCodec;
+import com.dianping.cat.message.spi.internal.MessageDispatcher;
 import com.site.lookup.configuration.AbstractResourceConfigurator;
 import com.site.lookup.configuration.Component;
 
@@ -21,7 +23,6 @@ public class ComponentsConfigurator extends AbstractResourceConfigurator {
 	@Override
 	public List<Component> defineComponents() {
 		List<Component> all = new ArrayList<Component>();
-
 		String inMemory = "in-memory";
 
 		all.add(C(InMemoryQueue.class));
@@ -32,6 +33,8 @@ public class ComponentsConfigurator extends AbstractResourceConfigurator {
 
 		all.add(C(MessageProducer.class, DefaultMessageProducer.class));
 
+		all.add(C(MessageCodec.class, "plain-text", PlainTextMessageCodec.class));
+
 		// the following are not used right now
 		all.add(C(MessageHandler.class, MessageDispatcher.class) //
 		      .req(MessageReceiver.class, inMemory));

cat-core/src/main/java/com/dianping/cat/message/broker/DefaultMessageBroker.java

 package com.dianping.cat.message.broker;
 
-import com.dianping.cat.message.Message;
-import com.dianping.cat.message.handler.MessageHandler;
 import com.dianping.cat.message.io.MessageReceiver;
 import com.dianping.cat.message.io.MessageSender;
+import com.dianping.cat.message.spi.MessageHandler;
+import com.dianping.cat.message.spi.MessageTree;
 import com.site.lookup.annotation.Inject;
 
 public class DefaultMessageBroker implements MessageBroker {
@@ -17,8 +17,8 @@ public class DefaultMessageBroker implements MessageBroker {
 	public void run() {
 		m_reciever.onMessage(new MessageHandler() {
 			@Override
-			public void handle(Message message) {
-				m_sender.send(message);
+			public void handle(MessageTree tree) {
+				m_sender.send(tree);
 			}
 		});
 	}

cat-core/src/main/java/com/dianping/cat/message/codec/MessageCodec.java

-package com.dianping.cat.message.codec;
-
-import com.dianping.cat.message.Message;
-
-public interface MessageCodec {
-	public byte[] encode(Message message);
-
-	public Message decode(byte[] data);
-}

cat-core/src/main/java/com/dianping/cat/message/consumer/MessageConsumer.java

-package com.dianping.cat.message.consumer;
-
-import com.dianping.cat.message.Message;
-
-public interface MessageConsumer {
-	public void consume(Message message);
-}

cat-core/src/main/java/com/dianping/cat/message/consumer/MessageConsumerRegistry.java

-package com.dianping.cat.message.consumer;
-
-import java.util.ArrayList;
-import java.util.List;
-
-public class MessageConsumerRegistry {
-	private List<MessageConsumer> m_consumers = new ArrayList<MessageConsumer>();
-
-	public List<MessageConsumer> getConsumers() {
-		return m_consumers;
-	}
-	
-	public void register(MessageConsumer consumer) {
-		m_consumers.add(consumer);
-	}
-}

cat-core/src/main/java/com/dianping/cat/message/handler/MessageHandler.java

-package com.dianping.cat.message.handler;
-
-import com.dianping.cat.message.Message;
-
-public interface MessageHandler {
-	public void handle(Message message);
-}

cat-core/src/main/java/com/dianping/cat/message/internal/DefaultMessageProducer.java

 package com.dianping.cat.message.internal;
 
+import org.codehaus.plexus.personality.plexus.lifecycle.phase.Initializable;
+import org.codehaus.plexus.personality.plexus.lifecycle.phase.InitializationException;
+
 import com.dianping.cat.message.Event;
 import com.dianping.cat.message.Heartbeat;
 import com.dianping.cat.message.Message;
 import com.dianping.cat.message.MessageProducer;
 import com.dianping.cat.message.Transaction;
+import com.dianping.cat.message.io.MessageSender;
+import com.site.lookup.annotation.Inject;
+
+public class DefaultMessageProducer implements MessageProducer, Initializable {
+	@Inject
+	private MessageSender m_sender;
+
+	@Override
+	public void initialize() throws InitializationException {
+		MessageManager.INSTANCE.initialize(m_sender);
+	}
 
-public class DefaultMessageProducer implements MessageProducer {
 	@Override
 	public void logEvent(String type, String name, String status, String nameValuePairs) {
 		Event event = newEvent(type, name);

cat-core/src/main/java/com/dianping/cat/message/internal/MessageManager.java

@@ -6,6 +6,10 @@ import java.util.Stack;
 
 import com.dianping.cat.message.Message;
 import com.dianping.cat.message.Transaction;
+import com.dianping.cat.message.io.MessageSender;
+import com.dianping.cat.message.spi.MessageTree;
+import com.dianping.cat.message.spi.internal.DefaultMessageTree;
+import com.site.lookup.annotation.Inject;
 
 public enum MessageManager {
 	INSTANCE;
@@ -17,42 +21,55 @@ public enum MessageManager {
 		}
 	};
 
+	@Inject
+	private MessageSender m_sender;
+
+	private boolean m_initialized;
+
 	public void add(Message message) {
-		s_context.get().add(message);
+		s_context.get().add(this, message);
 	}
 
 	public void end(Transaction transaction) {
-		s_context.get().end(transaction);
+		s_context.get().end(this, transaction);
 	}
 
-	public void start(Transaction transaction) {
-		s_context.get().start(transaction);
+	void flush(MessageTree tree) {
+		if (m_sender != null) {
+			m_sender.send(tree);
+			
+			// destroy current thread data
+			s_context.remove();
+		} else if (!m_initialized) {
+			throw new IllegalStateException("MessageManager is not initialized yet!");
+		}
 	}
 
-	void handle(Transaction transaction) {
-		// TODO
-		System.out.println(transaction);
+	public void initialize(MessageSender sender) {
+		m_sender = sender;
+		m_initialized = true;
+	}
+
+	public void start(Transaction transaction) {
+		s_context.get().start(transaction);
 	}
 
 	static class Context {
+		private DefaultMessageTree m_tree = new DefaultMessageTree();
+
 		private Stack<Transaction> m_stack = new Stack<Transaction>();
 
-		public void add(Message message) {
-			if (!m_stack.isEmpty()) {
+		public void add(MessageManager manager, Message message) {
+			if (m_stack.isEmpty()) {
+				m_tree.setMessage(message);
+			} else {
 				Transaction entry = m_stack.peek();
 
 				entry.addChild(message);
-			} else {
-				// add a mock transaction as its parent
-				Transaction t = new FakeTransaction();
-
-				start(t);
-				t.addChild(message);
-				end(t);
 			}
 		}
 
-		public void end(Transaction transaction) {
+		public void end(MessageManager manager, Transaction transaction) {
 			if (!m_stack.isEmpty()) {
 				Transaction current = m_stack.peek();
 
@@ -65,7 +82,7 @@ public enum MessageManager {
 				m_stack.pop();
 
 				if (m_stack.isEmpty()) {
-					INSTANCE.handle(transaction);
+					manager.flush(m_tree);
 				}
 			}
 		}
@@ -111,12 +128,11 @@ public enum MessageManager {
 		}
 
 		@Override
-		public void complete() {
+		public void addChild(Message message) {
 		}
 
 		@Override
-		public long getDuration() {
-			return -1;
+		public void complete() {
 		}
 
 		@Override
@@ -125,7 +141,8 @@ public enum MessageManager {
 		}
 
 		@Override
-		public void addChild(Message message) {
+		public long getDuration() {
+			return -1;
 		}
 	}
 }

cat-core/src/main/java/com/dianping/cat/message/io/InMemoryQueue.java

@@ -9,10 +9,10 @@ import org.codehaus.plexus.logging.Logger;
 import org.codehaus.plexus.personality.plexus.lifecycle.phase.Initializable;
 import org.codehaus.plexus.personality.plexus.lifecycle.phase.InitializationException;
 
-import com.dianping.cat.message.Message;
+import com.dianping.cat.message.spi.MessageTree;
 
 public class InMemoryQueue implements LogEnabled, Initializable {
-	private BlockingQueue<Message> m_queue;
+	private BlockingQueue<MessageTree> m_queue;
 
 	private int m_queueSize;
 
@@ -26,30 +26,42 @@ public class InMemoryQueue implements LogEnabled, Initializable {
 	@Override
 	public void initialize() throws InitializationException {
 		if (m_queueSize <= 0) {
-			m_queue = new LinkedBlockingQueue<Message>();
+			m_queue = new LinkedBlockingQueue<MessageTree>();
 		} else {
-			m_queue = new LinkedBlockingQueue<Message>(m_queueSize);
+			m_queue = new LinkedBlockingQueue<MessageTree>(m_queueSize);
 		}
 	}
 
-	public void offer(Message message) {
-		while (!m_queue.offer(message)) {
-			// throw away the message at the tail
-			Message m = m_queue.poll();
+	public void offer(MessageTree tree) {
+		while (!m_queue.offer(tree)) {
+			// throw away the tree at the tail
+			MessageTree m = m_queue.poll();
 
 			if (m == null) {
 				break;
 			} else {
-				m_logger.warn(message + " was thrown away due to queue is full!");
+				m_logger.warn(tree + " was thrown away due to queue is full!");
 			}
 		}
 	}
 
-	public Message poll(long timeout) throws InterruptedException {
-		return m_queue.poll(timeout, TimeUnit.MILLISECONDS);
+	public MessageTree peek() {
+	   return m_queue.peek();
+   }
+
+	public MessageTree poll(long timeout) throws InterruptedException {
+		if (timeout <= 0) {
+			return m_queue.poll();
+		} else {
+			return m_queue.poll(timeout, TimeUnit.MILLISECONDS);
+		}
 	}
 
 	public void setQueueSize(int queueSize) {
 		m_queueSize = queueSize;
 	}
+
+	public int size() {
+		return m_queue.size();
+	}
 }

cat-core/src/main/java/com/dianping/cat/message/io/InMemoryReceiver.java

 package com.dianping.cat.message.io;
 
-import com.dianping.cat.message.Message;
-import com.dianping.cat.message.handler.MessageHandler;
+import com.dianping.cat.message.spi.MessageHandler;
+import com.dianping.cat.message.spi.MessageTree;
 import com.site.lookup.annotation.Inject;
 
 public class InMemoryReceiver implements MessageReceiver {
@@ -24,10 +24,10 @@ public class InMemoryReceiver implements MessageReceiver {
 	public void onMessage(MessageHandler handler) {
 		try {
 			while (true) {
-				Message m = m_queue.poll(1);
+				MessageTree tree = m_queue.poll(1);
 
-				if (m != null) {
-					handler.handle(m);
+				if (tree != null) {
+					handler.handle(tree);
 				} else if (!isActive()) {
 					break;
 				}

cat-core/src/main/java/com/dianping/cat/message/io/InMemorySender.java

 package com.dianping.cat.message.io;
 
-import com.dianping.cat.message.Message;
+import com.dianping.cat.message.spi.MessageTree;
 import com.site.lookup.annotation.Inject;
 
 public class InMemorySender implements MessageSender {
@@ -20,9 +20,9 @@ public class InMemorySender implements MessageSender {
 	}
 
 	@Override
-	public void send(Message message) {
+	public void send(MessageTree tree) {
 		if (isActive()) {
-			m_queue.offer(message);
+			m_queue.offer(tree);
 		}
 	}
 

cat-core/src/main/java/com/dianping/cat/message/io/MessageReceiver.java

 package com.dianping.cat.message.io;
 
-import com.dianping.cat.message.handler.MessageHandler;
+import com.dianping.cat.message.spi.MessageHandler;
 
 public interface MessageReceiver {
 	public void initialize();

cat-core/src/main/java/com/dianping/cat/message/io/MessageSender.java

 package com.dianping.cat.message.io;
 
-import com.dianping.cat.message.Message;
+import com.dianping.cat.message.spi.MessageTree;
 
 public interface MessageSender {
 	public void initialize();
 
-	public void send(Message message);
+	public void send(MessageTree tree);
 
 	public void shutdown();
 }

cat-core/src/main/java/com/dianping/cat/message/io/TcpSocketReceiver.java

@@ -21,9 +21,9 @@ import org.jboss.netty.channel.group.DefaultChannelGroup;
 import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
 import org.jboss.netty.handler.codec.frame.FrameDecoder;
 
-import com.dianping.cat.message.Message;
-import com.dianping.cat.message.codec.MessageCodec;
-import com.dianping.cat.message.handler.MessageHandler;
+import com.dianping.cat.message.spi.MessageCodec;
+import com.dianping.cat.message.spi.MessageHandler;
+import com.dianping.cat.message.spi.MessageTree;
 import com.site.lookup.annotation.Inject;
 
 public class TcpSocketReceiver implements MessageReceiver {
@@ -43,9 +43,9 @@ public class TcpSocketReceiver implements MessageReceiver {
 	private MessageHandler m_messageHandler;
 
 	void handleMessage(byte[] data) {
-		Message message = m_codec.decode(data);
+		MessageTree tree = m_codec.decode(data);
 
-		m_messageHandler.handle(message);
+		m_messageHandler.handle(tree);
 	}
 
 	@Override

cat-core/src/main/java/com/dianping/cat/message/io/TcpSocketSender.java

@@ -18,8 +18,8 @@ import org.jboss.netty.channel.MessageEvent;
 import org.jboss.netty.channel.SimpleChannelHandler;
 import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
 
-import com.dianping.cat.message.Message;
-import com.dianping.cat.message.codec.MessageCodec;
+import com.dianping.cat.message.spi.MessageCodec;
+import com.dianping.cat.message.spi.MessageTree;
 import com.site.lookup.annotation.Inject;
 
 public class TcpSocketSender implements MessageSender {
@@ -69,14 +69,15 @@ public class TcpSocketSender implements MessageSender {
 	}
 
 	@Override
-	public void send(Message message) {
-		byte[] data = m_codec.encode(message);
+	public void send(MessageTree tree) {
+		byte[] data = m_codec.encode(tree);
 		Channel channel = m_future.getChannel();
 		ChannelBufferFactory factory = channel.getConfig().getBufferFactory();
 		ChannelBuffer buf = factory.getBuffer(data.length + 4);
 
 		buf.writeInt(data.length);
 		buf.writeBytes(data);
+		
 		channel.write(buf);
 	}
 

cat-core/src/main/java/com/dianping/cat/message/io/Transport.java

 package com.dianping.cat.message.io;
 
 import com.dianping.cat.message.Message;
-import com.dianping.cat.message.handler.MessageHandler;
+import com.dianping.cat.message.spi.MessageHandler;
 
 /**
  * <p>

cat-core/src/main/java/com/dianping/cat/message/spi/AbstractMessageAnalyzer.java

+package com.dianping.cat.message.spi;
+
+public abstract class AbstractMessageAnalyzer<R> implements MessageAnalyzer {
+	@Override
+	public void analyze(MessageQueue queue) {
+		while (queue.isActive()) {
+			MessageTree tree = queue.poll();
+
+			if (tree != null) {
+				process(tree);
+			}
+		}
+
+		R result = generate();
+
+		store(result);
+	}
+
+	protected abstract void store(R result);
+
+	public abstract R generate();
+
+	protected abstract void process(MessageTree tree);
+}

cat-core/src/main/java/com/dianping/cat/message/spi/MessageAnalyzer.java

+package com.dianping.cat.message.spi;
+
+public interface MessageAnalyzer {
+	public void analyze(MessageQueue queue);
+}

cat-core/src/main/java/com/dianping/cat/message/spi/MessageCodec.java

+package com.dianping.cat.message.spi;
+
+public interface MessageCodec {
+	public byte[] encode(MessageTree tree);
+
+	public MessageTree decode(byte[] data);
+}

cat-core/src/main/java/com/dianping/cat/message/spi/MessageConsumer.java

+package com.dianping.cat.message.spi;
+
+
+public interface MessageConsumer {
+	public String getId();
+
+	public void consume(MessageTree tree);
+}

cat-core/src/main/java/com/dianping/cat/message/spi/MessageConsumerRegistry.java

+package com.dianping.cat.message.spi;
+
+import java.util.Map;
+
+public interface MessageConsumerRegistry {
+	public void registerFilter(MessageFilter filter);
+
+	public void registerConsumer(MessageConsumer consumer);
+
+	public Map<String, MessageConsumer> getConsumers();
+}
\ No newline at end of file

cat-core/src/main/java/com/dianping/cat/message/spi/MessageFilter.java

+package com.dianping.cat.message.spi;
+
+public interface MessageFilter {
+	public String getConsumerId();
+
+	public boolean doFilter(byte[] data);
+}

cat-core/src/main/java/com/dianping/cat/message/spi/MessageHandler.java

+package com.dianping.cat.message.spi;
+
+public interface MessageHandler {
+	public void handle(MessageTree message);
+}

cat-core/src/main/java/com/dianping/cat/message/spi/MessageQueue.java

+package com.dianping.cat.message.spi;
+
+public interface MessageQueue {
+	public boolean isActive();
+
+	public MessageTree poll();
+}

cat-core/src/main/java/com/dianping/cat/message/spi/MessageTree.java

+package com.dianping.cat.message.spi;
+
+public interface MessageTree {
+	public String getDomain();
+
+	public String getHostName();
+
+	public String getIpAddress();
+
+	public MessageTree getMessage();
+
+	public String getMessageId();
+
+	public int getPort();
+
+	public String getRequestToken();
+
+	public String getSessionToken();
+
+	public String getThreadId();
+}

cat-core/src/main/java/com/dianping/cat/message/spi/codec/PlainTextMessageCodec.java

+package com.dianping.cat.message.spi.codec;
+
+import com.dianping.cat.message.spi.MessageCodec;
+import com.dianping.cat.message.spi.MessageTree;
+
+public class PlainTextMessageCodec implements MessageCodec {
+	@Override
+	public byte[] encode(MessageTree tree) {
+		return null;
+	}
+
+	@Override
+	public MessageTree decode(byte[] data) {
+		return null;
+	}
+}

cat-core/src/main/java/com/dianping/cat/message/spi/internal/DefaultMessageConsumerRegistry.java

+package com.dianping.cat.message.spi.internal;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import com.dianping.cat.message.spi.MessageConsumer;
+import com.dianping.cat.message.spi.MessageConsumerRegistry;
+import com.dianping.cat.message.spi.MessageFilter;
+
+public class DefaultMessageConsumerRegistry implements MessageConsumerRegistry {
+	private Map<String, MessageConsumer> m_consumers = new LinkedHashMap<String, MessageConsumer>();
+
+	@Override
+	public Map<String, MessageConsumer> getConsumers() {
+		return m_consumers;
+	}
+
+	@Override
+	public void registerConsumer(MessageConsumer consumer) {
+		m_consumers.put(consumer.getId(), consumer);
+	}
+
+	@Override
+	public void registerFilter(MessageFilter filter) {
+
+	}
+
+	static class Entry {
+		private MessageConsumer m_message;
+
+		private List<MessageFilter> m_filters;
+	}
+}

cat-core/src/main/java/com/dianping/cat/message/spi/internal/DefaultMessageTree.java

+package com.dianping.cat.message.spi.internal;
+
+import com.dianping.cat.message.Message;
+import com.dianping.cat.message.spi.MessageTree;
+
+public class DefaultMessageTree implements MessageTree {
+	@Override
+	public String getDomain() {
+		return null;
+	}
+
+	@Override
+	public String getHostName() {
+		return null;
+	}
+
+	@Override
+	public String getIpAddress() {
+		return null;
+	}
+
+	@Override
+	public MessageTree getMessage() {
+		return null;
+	}
+
+	@Override
+	public String getMessageId() {
+		return null;
+	}
+
+	@Override
+	public int getPort() {
+		return 0;
+	}
+
+	@Override
+	public String getRequestToken() {
+		return null;
+	}
+
+	@Override
+	public String getSessionToken() {
+		return null;
+	}
+
+	@Override
+	public String getThreadId() {
+		return null;
+	}
+
+	public void setMessage(Message message) {
+	   
+   }
+}

cat-core/src/main/java/com/dianping/cat/message/handler/MessageDispatcher.java → cat-core/src/main/java/com/dianping/cat/message/spi/internal/MessageDispatcher.java

-package com.dianping.cat.message.handler;
+package com.dianping.cat.message.spi.internal;
 
-import java.util.List;
+import java.util.Collection;
 
-import com.dianping.cat.message.Message;
-import com.dianping.cat.message.consumer.MessageConsumer;
-import com.dianping.cat.message.consumer.MessageConsumerRegistry;
 import com.dianping.cat.message.io.MessageReceiver;
+import com.dianping.cat.message.spi.MessageConsumer;
+import com.dianping.cat.message.spi.MessageHandler;
+import com.dianping.cat.message.spi.MessageTree;
 import com.site.lookup.annotation.Inject;
 
 public class MessageDispatcher implements MessageHandler, Runnable {
@@ -13,15 +13,15 @@ public class MessageDispatcher implements MessageHandler, Runnable {
 	private MessageReceiver m_receiver;
 
 	@Inject
-	private MessageConsumerRegistry m_registry;
+	private DefaultMessageConsumerRegistry m_registry;
 
 	@Override
-	public void handle(Message message) {
-		List<MessageConsumer> comsumers = m_registry.getConsumers();
+	public void handle(MessageTree tree) {
+		Collection<MessageConsumer> comsumers = m_registry.getConsumers().values();
 
 		for (MessageConsumer consumer : comsumers) {
 			try {
-				consumer.consume(message);
+				consumer.consume(tree);
 			} catch (Exception e) {
 				e.printStackTrace();
 			}
@@ -37,7 +37,7 @@ public class MessageDispatcher implements MessageHandler, Runnable {
 		m_receiver = receiver;
 	}
 
-	public void setRegistry(MessageConsumerRegistry registry) {
+	public void setRegistry(DefaultMessageConsumerRegistry registry) {
 		m_registry = registry;
 	}
 }

cat-core/src/main/resources/META-INF/plexus/components.xml

@@ -29,8 +29,13 @@
 			<implementation>com.dianping.cat.message.internal.DefaultMessageProducer</implementation>
 		</component>
 		<component>
-			<role>com.dianping.cat.message.handler.MessageHandler</role>
-			<implementation>com.dianping.cat.message.handler.MessageDispatcher</implementation>
+			<role>com.dianping.cat.message.spi.MessageCodec</role>
+			<role-hint>plain-text</role-hint>
+			<implementation>com.dianping.cat.message.spi.codec.PlainTextMessageCodec</implementation>
+		</component>
+		<component>
+			<role>com.dianping.cat.message.spi.MessageHandler</role>
+			<implementation>com.dianping.cat.message.spi.internal.MessageDispatcher</implementation>
 			<requirements>
 				<requirement>
 					<role>com.dianping.cat.message.io.MessageReceiver</role>

cat-core/src/test/java/com/dianping/cat/message/internal/MessageProducerTest.java

+package com.dianping.cat.message.internal;
+
+import junit.framework.Assert;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import com.dianping.cat.message.Message;
+import com.dianping.cat.message.MessageProducer;
+import com.dianping.cat.message.Transaction;
+import com.dianping.cat.message.io.InMemoryQueue;
+import com.dianping.cat.message.spi.MessageCodec;
+import com.dianping.cat.message.spi.MessageTree;
+import com.site.lookup.ComponentTestCase;
+
+@RunWith(JUnit4.class)
+public class MessageProducerTest extends ComponentTestCase {
+	@Test
+	public void testNormal() throws Exception {
+		MessageProducer producer = lookup(MessageProducer.class);
+		InMemoryQueue queue = lookup(InMemoryQueue.class);
+		MessageCodec codec = lookup(MessageCodec.class, "plain-text");
+		Transaction t = producer.newTransaction("URL", "MyPage");
+
+		try {
+			// do your business here
+			t.addData("k1", "v1");
+			t.addData("k2", "v2");
+			t.addData("k3", "v3");
+			Thread.sleep(30);
+
+			producer.logEvent("URL", "Payload", Message.SUCCESS, "host=my-host&ip=127.0.0.1&agent=...");
+			t.setStatus(Message.SUCCESS);
+		} catch (Exception e) {
+			t.setStatus(e);
+		} finally {
+			t.complete();
+		}
+
+		Assert.assertEquals("One message should be in the queue.", 1, queue.size());
+
+		MessageTree tree = queue.peek();
+
+		Assert.assertEquals("...", codec.encode(tree));
+	}
+}

cat-core/src/test/java/com/dianping/cat/message/internal/MessageProducerTestConfigurator.java

+package com.dianping.cat.message.internal;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+
+import com.dianping.cat.message.MessageProducer;
+import com.dianping.cat.message.io.MessageSender;
+import com.site.lookup.configuration.AbstractResourceConfigurator;
+import com.site.lookup.configuration.Component;
+
+public class MessageProducerTestConfigurator extends AbstractResourceConfigurator {
+	public static void main(String[] args) {
+		generatePlexusComponentsXmlFile(new MessageProducerTestConfigurator());
+	}
+
+	@Override
+	public List<Component> defineComponents() {
+		List<Component> all = new ArrayList<Component>();
+		String inMemory = "in-memory";
+
+		all.add(C(MessageProducer.class, DefaultMessageProducer.class) //
+		      .req(MessageSender.class, inMemory));
+
+		return all;
+	}
+
+	@Override
+	protected File getConfigurationFile() {
+		return new File("src/test/resources/" + MessageProducerTest.class.getName().replace('.', '/') + ".xml");
+	}
+}

cat-core/src/test/java/com/dianping/cat/message/io/InMemoryTest.java

@@ -12,9 +12,9 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
-import com.dianping.cat.message.Message;
-import com.dianping.cat.message.handler.MessageHandler;
-import com.dianping.cat.message.internal.AbstractMessage;
+import com.dianping.cat.message.spi.MessageHandler;
+import com.dianping.cat.message.spi.MessageTree;
+import com.dianping.cat.message.spi.internal.DefaultMessageTree;
 import com.site.lookup.ComponentTestCase;
 
 @RunWith(JUnit4.class)
@@ -41,7 +41,7 @@ public class InMemoryTest extends ComponentTestCase {
 				sender.initialize();
 
 				for (int i = 0; i < len; i++) {
-					sender.send(new MockMessage());
+					sender.send(new DefaultMessageTree());
 				}
 
 				sender.shutdown();
@@ -58,16 +58,6 @@ public class InMemoryTest extends ComponentTestCase {
 		Assert.assertEquals(len, sb.length());
 	}
 
-	static class MockMessage extends AbstractMessage {
-		public MockMessage() {
-			super(null, null);
-		}
-
-		@Override
-		public void complete() {
-		}
-	}
-
 	static class MockMessageHandler implements MessageHandler {
 		private StringBuilder m_sb;
 
@@ -76,7 +66,7 @@ public class InMemoryTest extends ComponentTestCase {
 		}
 
 		@Override
-		public void handle(Message message) {
+		public void handle(MessageTree tree) {
 			m_sb.append('.');
 		}
 	}

cat-core/src/test/java/com/dianping/cat/message/io/TcpSocketTest.java

@@ -12,10 +12,11 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
-import com.dianping.cat.message.Message;
-import com.dianping.cat.message.codec.MessageCodec;
-import com.dianping.cat.message.handler.MessageHandler;
 import com.dianping.cat.message.internal.AbstractMessage;
+import com.dianping.cat.message.spi.MessageCodec;
+import com.dianping.cat.message.spi.MessageHandler;
+import com.dianping.cat.message.spi.MessageTree;
+import com.dianping.cat.message.spi.internal.DefaultMessageTree;
 import com.site.lookup.ComponentTestCase;
 
 @RunWith(JUnit4.class)
@@ -42,7 +43,7 @@ public class TcpSocketTest extends ComponentTestCase {
 				sender.initialize();
 
 				for (int i = 0; i < len; i++) {
-					sender.send(new MockMessage());
+					sender.send(new DefaultMessageTree());
 				}
 			}
 		}));
@@ -72,12 +73,12 @@ public class TcpSocketTest extends ComponentTestCase {
 
 	public static class MockMessageCodec implements MessageCodec {
 		@Override
-		public Message decode(byte[] bytes) {
-			return new MockMessage();
+		public MessageTree decode(byte[] bytes) {
+			return new DefaultMessageTree();
 		}
 
 		@Override
-		public byte[] encode(Message message) {
+		public byte[] encode(MessageTree message) {
 			return "mock".getBytes();
 		}
 	}
@@ -90,7 +91,7 @@ public class TcpSocketTest extends ComponentTestCase {
 		}
 
 		@Override
-		public void handle(Message message) {
+		public void handle(MessageTree tree) {
 			m_sb.append('.');
 		}
 	}

cat-core/src/test/java/com/dianping/cat/message/io/TcpSocketTestConfigurator.java

@@ -4,8 +4,8 @@ import java.io.File;
 import java.util.ArrayList;
 import java.util.List;
 
-import com.dianping.cat.message.codec.MessageCodec;
 import com.dianping.cat.message.io.TcpSocketTest.MockMessageCodec;
+import com.dianping.cat.message.spi.MessageCodec;
 import com.site.lookup.configuration.AbstractResourceConfigurator;
 import com.site.lookup.configuration.Component;
 

cat-core/src/test/java/com/dianping/cat/message/io/TransportManagerTest.java

@@ -5,9 +5,9 @@ import junit.framework.Assert;
 import org.junit.Test;
 
 import com.dianping.cat.message.Message;
-import com.dianping.cat.message.handler.MessageHandler;
 import com.dianping.cat.message.io.Transport;
 import com.dianping.cat.message.io.TransportManager;
+import com.dianping.cat.message.spi.MessageHandler;
 
 public class TransportManagerTest {
    @Test

cat-core/src/test/resources/com/dianping/cat/message/internal/MessageProducerTest.xml

+<plexus>
+	<components>
+		<component>
+			<role>com.dianping.cat.message.MessageProducer</role>
+			<implementation>com.dianping.cat.message.internal.DefaultMessageProducer</implementation>
+			<requirements>
+				<requirement>
+					<role>com.dianping.cat.message.io.MessageSender</role>
+					<role-hint>in-memory</role-hint>
+				</requirement>
+			</requirements>
+		</component>
+	</components>
+</plexus>

cat-core/src/test/resources/com/dianping/cat/message/io/TcpSocketTest.xml

 <plexus>
 	<components>
 		<component>
-			<role>com.dianping.cat.message.codec.MessageCodec</role>
+			<role>com.dianping.cat.message.spi.MessageCodec</role>
 			<role-hint>tcp-socket</role-hint>
 			<implementation>com.dianping.cat.message.io.TcpSocketTest$MockMessageCodec</implementation>
 		</component>
@@ -14,7 +14,7 @@
 			</configuration>
 			<requirements>
 				<requirement>
-					<role>com.dianping.cat.message.codec.MessageCodec</role>
+					<role>com.dianping.cat.message.spi.MessageCodec</role>
 					<role-hint>tcp-socket</role-hint>
 				</requirement>
 			</requirements>
@@ -28,7 +28,7 @@
 			</configuration>
 			<requirements>
 				<requirement>
-					<role>com.dianping.cat.message.codec.MessageCodec</role>
+					<role>com.dianping.cat.message.spi.MessageCodec</role>
 					<role-hint>tcp-socket</role-hint>
 				</requirement>
 			</requirements>