rename transport to io package

cat-core/pom.xml

@@ -18,5 +18,9 @@
 			<groupId>com.site.common</groupId>
 			<artifactId>lookup</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>org.jboss.netty</groupId>
+			<artifactId>netty</artifactId>
+		</dependency>
 	</dependencies>
 </project>

cat-core/src/main/java/com/dianping/cat/Cat.java

 package com.dianping.cat;
 
-import com.dianping.cat.message.MessageFactory;
+import com.dianping.cat.message.MessageProducer;
 
 
 /**
@@ -9,7 +9,7 @@ import com.dianping.cat.message.MessageFactory;
  * @author Frankie Wu
  */
 public class Cat {
-	public static MessageFactory getFactory() {
+	public static MessageProducer getProducer() {
 		return null;
 	}
 }

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

@@ -3,30 +3,41 @@ package com.dianping.cat.configuration;
 import java.util.ArrayList;
 import java.util.List;
 
-import com.dianping.cat.message.transport.InMemoryTransport;
-import com.dianping.cat.message.transport.TcpSocketTransport;
-import com.dianping.cat.message.transport.Transport;
-import com.dianping.cat.message.transport.TransportManager;
-import com.dianping.cat.message.transport.UdpMulticastTransport;
+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.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.site.lookup.configuration.AbstractResourceConfigurator;
 import com.site.lookup.configuration.Component;
 
 public class ComponentsConfigurator extends AbstractResourceConfigurator {
-   @Override
-   public List<Component> defineComponents() {
-      List<Component> all = new ArrayList<Component>();
+	@Override
+	public List<Component> defineComponents() {
+		List<Component> all = new ArrayList<Component>();
 
-      all.add(C(Transport.class, "in-memory", InMemoryTransport.class));
-      all.add(C(Transport.class, "tcp-socket", TcpSocketTransport.class));
-      all.add(C(Transport.class, "udp-multicast", UdpMulticastTransport.class));
+		String inMemory = "in-memory";
 
-      all.add(C(TransportManager.class) //
-            .req(Transport.class, "in-memory"));
+		all.add(C(InMemoryQueue.class));
+		all.add(C(MessageSender.class, inMemory, InMemorySender.class) //
+		      .req(InMemoryQueue.class));
+		all.add(C(MessageReceiver.class, inMemory, InMemoryReceiver.class) //
+		      .req(InMemoryQueue.class));
+		all.add(C(MessageHandler.class, MessageDispatcher.class) //
+		      .req(MessageReceiver.class, inMemory));
 
-      return all;
-   }
+		all.add(C(MessageBroker.class, inMemory, DefaultMessageBroker.class) //
+		      .req(MessageSender.class, inMemory) //
+		      .req(MessageReceiver.class, inMemory));
 
-   public static void main(String[] args) {
-      generatePlexusComponentsXmlFile(new ComponentsConfigurator());
-   }
+		return all;
+	}
+
+	public static void main(String[] args) {
+		generatePlexusComponentsXmlFile(new ComponentsConfigurator());
+	}
 }

cat-core/src/main/java/com/dianping/cat/message/Message.java

@@ -15,70 +15,83 @@ package com.dianping.cat.message;
  * @author Frankie Wu
  */
 public interface Message {
-   public static final String SUCCESS = "0";
+	public static final String SUCCESS = "0";
 
-   /**
-    * add one or multiple key-value pairs to the message.
-    * 
-    * @param keyValuePairs
-    *           key-value pairs like 'a=1&b=2&...'
-    */
-   public void addData(String keyValuePairs);
+	/**
+	 * add one or multiple key-value pairs to the message.
+	 * 
+	 * @param keyValuePairs
+	 *           key-value pairs like 'a=1&b=2&...'
+	 */
+	public void addData(String keyValuePairs);
 
-   /**
-    * add one key-value pair to the message.
-    * 
-    * @param key
-    * @param value
-    */
-   public void addData(String key, Object value);
+	/**
+	 * add one key-value pair to the message.
+	 * 
+	 * @param key
+	 * @param value
+	 */
+	public void addData(String key, Object value);
 
-   /**
-    * Complete the message construction.
-    */
-   public void complete();
+	/**
+	 * Complete the message construction.
+	 */
+	public void complete();
 
-   /**
-    * Message name.
-    * 
-    * @return message name
-    */
-   public String getName();
+	/**
+	 * Message name.
+	 * 
+	 * @return message name
+	 */
+	public String getName();
 
-   /**
-    * Get the message status.
-    * 
-    * @return message status. "0" means success, otherwise error code.
-    */
-   public String getStatus();
+	/**
+	 * Get the message status.
+	 * 
+	 * @return message status. "0" means success, otherwise error code.
+	 */
+	public String getStatus();
 
-   /**
-    * The time stamp the message was created.
-    * 
-    * @return message creation time stamp in milliseconds
-    */
-   public long getTimestamp();
+	/**
+	 * The time stamp the message was created.
+	 * 
+	 * @return message creation time stamp in milliseconds
+	 */
+	public long getTimestamp();
 
-   /**
-    * Message type.
-    * 
-    * @return message type
-    */
-   public String getType();
+	/**
+	 * Message type.
+	 * 
+	 * <p>
+	 * Typical message types are:
+	 * <ul>
+	 * <li>URL: maps to one method of an action</li>
+	 * <li>Service: maps to one method of service call</li>
+	 * <li>Search: maps to one method of search call</li>
+	 * <li>SQL: maps to one SQL statement</li>
+	 * <li>Cache: maps to one cache access</li>
+	 * <li>Exception: maps to java.lang.Exception</li>
+	 * <li>Error: maps to java.lang.Error</li>
+	 * </ul>
+	 * </p>
+	 * 
+	 * @return message type
+	 */
+	public String getType();
 
-   /**
-    * Set the message status.
-    * 
-    * @param status
-    *           message status. "0" means success, otherwise error code.
-    */
-   public void setStatus(String status);
+	/**
+	 * Set the message status.
+	 * 
+	 * @param status
+	 *           message status. "0" means success, otherwise error code.
+	 */
+	public void setStatus(String status);
 
-   /**
-    * Set the message status with exception class name.
-    * 
-    * @param e
-    *           exception.
-    */
-   public void setStatus(Throwable e);
+	/**
+	 * Set the message status with exception class name.
+	 * 
+	 * @param e
+	 *           exception.
+	 */
+	public void setStatus(Throwable e);
 }

cat-core/src/main/java/com/dianping/cat/message/MessageFactory.java → cat-core/src/main/java/com/dianping/cat/message/MessageProducer.java

@@ -117,7 +117,7 @@ package com.dianping.cat.message;
  * 
  * @author Frankie Wu
  */
-public interface MessageFactory {
+public interface MessageProducer {
 	/**
 	 * Log an event in one shot.
 	 * 

cat-core/src/main/java/com/dianping/cat/message/Transaction.java

@@ -34,5 +34,5 @@ package com.dianping.cat.message;
  * @author Frankie Wu
  */
 public interface Transaction extends Message {
-   public long getDuration();
+	public long getDuration();
 }

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.site.lookup.annotation.Inject;
+
+public class DefaultMessageBroker implements MessageBroker {
+	@Inject
+	private MessageReceiver m_reciever;
+
+	@Inject
+	private MessageSender m_sender;
+
+	@Override
+	public void run() {
+		m_reciever.onMessage(new MessageHandler() {
+			@Override
+			public void handle(Message message) {
+				m_sender.send(message);
+			}
+		});
+	}
+
+	public void setReciever(MessageReceiver reciever) {
+		m_reciever = reciever;
+	}
+
+	public void setSender(MessageSender sender) {
+		m_sender = sender;
+	}
+}

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

@@ -3,5 +3,5 @@ package com.dianping.cat.message.consumer;
 import com.dianping.cat.message.Message;
 
 public interface MessageConsumer {
-   public void consume(Message message);
+	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/MessageDispatcher.java

@@ -4,11 +4,40 @@ import java.util.List;
 
 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.site.lookup.annotation.Inject;
 
-public class MessageDispatcher implements MessageHandler {
-   private List<MessageConsumer> m_comsumers;
+public class MessageDispatcher implements MessageHandler, Runnable {
+	@Inject
+	private MessageReceiver m_receiver;
 
-   @Override
-   public void handle(Message message) {
-   }
+	@Inject
+	private MessageConsumerRegistry m_registry;
+
+	@Override
+	public void handle(Message message) {
+		List<MessageConsumer> comsumers = m_registry.getConsumers();
+
+		for (MessageConsumer consumer : comsumers) {
+			try {
+				consumer.consume(message);
+			} catch (Exception e) {
+				e.printStackTrace();
+			}
+		}
+	}
+
+	@Override
+	public void run() {
+		m_receiver.onMessage(this);
+	}
+
+	public void setReceiver(MessageReceiver receiver) {
+		m_receiver = receiver;
+	}
+
+	public void setRegistry(MessageConsumerRegistry registry) {
+		m_registry = registry;
+	}
 }

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

+package com.dianping.cat.message.io;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.codehaus.plexus.logging.LogEnabled;
+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;
+
+public class InMemoryQueue implements LogEnabled, Initializable {
+	private BlockingQueue<Message> m_queue;
+
+	private int m_queueSize;
+
+	private Logger m_logger;
+
+	@Override
+	public void enableLogging(Logger logger) {
+		m_logger = logger;
+	}
+
+	@Override
+	public void initialize() throws InitializationException {
+		if (m_queueSize <= 0) {
+			m_queue = new LinkedBlockingQueue<Message>();
+		} else {
+			m_queue = new LinkedBlockingQueue<Message>(m_queueSize);
+		}
+	}
+
+	public void offer(Message message) {
+		while (!m_queue.offer(message)) {
+			// throw away the message at the tail
+			Message m = m_queue.poll();
+
+			if (m == null) {
+				break;
+			} else {
+				m_logger.warn(message + " was thrown away due to queue is full!");
+			}
+		}
+	}
+
+	public Message poll(long timeout) throws InterruptedException {
+		return m_queue.poll(timeout, TimeUnit.MILLISECONDS);
+	}
+
+	public void setQueueSize(int queueSize) {
+		m_queueSize = queueSize;
+	}
+}

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

-package com.dianping.cat.message.transport;
-
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
+package com.dianping.cat.message.io;
 
 import com.dianping.cat.message.Message;
 import com.dianping.cat.message.handler.MessageHandler;
+import com.site.lookup.annotation.Inject;
 
-public class InMemoryTransport implements Transport {
-	private BlockingQueue<Message> m_queue;
+public class InMemoryReceiver implements MessageReceiver {
+	@Inject
+	private InMemoryQueue m_queue;
 
 	private transient boolean m_active = true;
 
-	public InMemoryTransport() {
-		m_queue = new LinkedBlockingQueue<Message>();
-	}
-
-	public InMemoryTransport(int queueSize) {
-		m_queue = new LinkedBlockingQueue<Message>(queueSize);
+	@Override
+	public void initialize() {
 	}
 
-	@Override
-	public void send(Message message) {
-		while (m_active && !m_queue.offer(message)) {
-			// throw away the message at the tail
-			Message m = m_queue.poll();
-
-			if (m == null) {
-				break;
-			} else {
-				System.out.println(message + " was thrown away due to queue is full!");
-			}
+	public boolean isActive() {
+		synchronized (this) {
+			return m_active;
 		}
 	}
 
@@ -38,11 +24,11 @@ public class InMemoryTransport implements Transport {
 	public void onMessage(MessageHandler handler) {
 		try {
 			while (true) {
-				Message m = m_queue.poll(1L, TimeUnit.MILLISECONDS);
+				Message m = m_queue.poll(1);
 
 				if (m != null) {
 					handler.handle(m);
-				} else if (!m_active) {
+				} else if (!isActive()) {
 					break;
 				}
 			}
@@ -51,8 +37,14 @@ public class InMemoryTransport implements Transport {
 		}
 	}
 
+	public void setQueue(InMemoryQueue queue) {
+		m_queue = queue;
+	}
+
 	@Override
 	public void shutdown() {
-		m_active = false;
+		synchronized (this) {
+			m_active = false;
+		}
 	}
 }

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.site.lookup.annotation.Inject;
+
+public class InMemorySender implements MessageSender {
+	@Inject
+	private InMemoryQueue m_queue;
+
+	private transient boolean m_active = true;
+
+	@Override
+	public void initialize() {
+	}
+
+	public boolean isActive() {
+		synchronized (this) {
+			return m_active;
+		}
+	}
+
+	@Override
+	public void send(Message message) {
+		if (isActive()) {
+			m_queue.offer(message);
+		}
+	}
+
+	public void setQueue(InMemoryQueue queue) {
+		m_queue = queue;
+	}
+
+	@Override
+	public void shutdown() {
+		synchronized (this) {
+			m_active = false;
+		}
+	}
+}

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;
+
+public interface MessageReceiver {
+	public void initialize();
+
+	public void onMessage(MessageHandler handler);
+
+	public void shutdown();
+}

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

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

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

+package com.dianping.cat.message.io;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.Executors;
+
+import org.jboss.netty.bootstrap.ServerBootstrap;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFactory;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelPipeline;
+import org.jboss.netty.channel.ChannelPipelineFactory;
+import org.jboss.netty.channel.Channels;
+import org.jboss.netty.channel.ExceptionEvent;
+import org.jboss.netty.channel.MessageEvent;
+import org.jboss.netty.channel.SimpleChannelHandler;
+import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
+import org.jboss.netty.handler.codec.frame.FrameDecoder;
+
+import com.dianping.cat.message.handler.MessageHandler;
+import com.site.lookup.annotation.Inject;
+
+public class TcpSocketReceiver implements MessageReceiver {
+	@Inject
+	private String m_host;
+
+	@Inject
+	private int m_port = 2280; // default port number from phone, C:2, A:2, T:8
+
+	@Override
+	public void initialize() {
+		InetSocketAddress address;
+
+		if (m_host == null) {
+			address = new InetSocketAddress(m_port);
+		} else {
+			address = new InetSocketAddress(m_host, m_port);
+		}
+
+		ChannelFactory factory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
+		      Executors.newCachedThreadPool());
+		ServerBootstrap bootstrap = new ServerBootstrap(factory);
+
+		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
+			public ChannelPipeline getPipeline() {
+				return Channels.pipeline(new MyDecoder(), new MyHandler());
+			}
+		});
+
+		bootstrap.setOption("child.tcpNoDelay", true);
+		bootstrap.setOption("child.keepAlive", true);
+		bootstrap.bind(address);
+	}
+
+	public void setHost(String host) {
+		m_host = host;
+	}
+
+	public void setPort(int port) {
+		m_port = port;
+	}
+
+	@Override
+	public void onMessage(MessageHandler handler) {
+
+	}
+
+	@Override
+	public void shutdown() {
+
+	}
+
+	public class MyDecoder extends FrameDecoder {
+		@Override
+		protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) {
+			if (buffer.readableBytes() < 4) {
+				return null;
+			}
+
+			return buffer.readBytes(4);
+		}
+	}
+
+	public static class MyHandler extends SimpleChannelHandler {
+		@Override
+		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
+			ChannelBuffer buf = (ChannelBuffer) e.getMessage();
+
+			while (buf.readable()) {
+				System.out.println((char) buf.readByte());
+				System.out.flush();
+			}
+		}
+
+		@Override
+		public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
+			e.getCause().printStackTrace();
+
+			e.getChannel().close();
+		}
+	}
+}

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

+package com.dianping.cat.message.io;
+
+import java.net.InetSocketAddress;
+import java.util.concurrent.Executors;
+
+import org.jboss.netty.bootstrap.ClientBootstrap;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFactory;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelPipeline;
+import org.jboss.netty.channel.ChannelPipelineFactory;
+import org.jboss.netty.channel.ChannelStateEvent;
+import org.jboss.netty.channel.Channels;
+import org.jboss.netty.channel.ExceptionEvent;
+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.site.lookup.annotation.Inject;
+
+public class TcpSocketSender implements MessageSender {
+	@Inject
+	private String m_host;
+
+	@Inject
+	private int m_port = 2280; // default port number from phone, C:2, A:2, T:8
+
+	@Override
+	public void initialize() {
+		InetSocketAddress address = new InetSocketAddress(m_host, m_port);
+		ChannelFactory factory = new NioClientSocketChannelFactory(Executors.newCachedThreadPool(),
+		      Executors.newCachedThreadPool());
+		ClientBootstrap bootstrap = new ClientBootstrap(factory);
+
+		bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
+			public ChannelPipeline getPipeline() {
+				return Channels.pipeline(new MyHandler());
+			}
+		});
+
+		bootstrap.setOption("tcpNoDelay", true);
+		bootstrap.setOption("keepAlive", true);
+		bootstrap.connect(address);
+	}
+
+	public void setHost(String host) {
+		m_host = host;
+	}
+
+	public void setPort(int port) {
+		m_port = port;
+	}
+
+	@Override
+	public void shutdown() {
+
+	}
+
+	public static class MyHandler extends SimpleChannelHandler {
+		private Channel m_channel;
+
+		@Override
+		public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
+			m_channel = e.getChannel();
+		}
+
+		@Override
+		public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
+			ChannelBuffer buf = (ChannelBuffer) e.getMessage();
+
+			while (buf.readable()) {
+				System.out.println((char) buf.readByte());
+				System.out.flush();
+			}
+		}
+
+		@Override
+		public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
+			e.getCause().printStackTrace();
+
+			e.getChannel().close();
+		}
+	}
+
+	@Override
+	public void send(Message message) {
+		// TODO Auto-generated method stub
+
+	}
+}

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

-package com.dianping.cat.message.transport;
+package com.dianping.cat.message.io;
 
 import com.dianping.cat.message.Message;
 import com.dianping.cat.message.handler.MessageHandler;

cat-core/src/main/java/com/dianping/cat/message/transport/TransportManager.java → cat-core/src/main/java/com/dianping/cat/message/io/TransportManager.java

-package com.dianping.cat.message.transport;
+package com.dianping.cat.message.io;
 
 public class TransportManager {
    private static TransportManager s_instance;

cat-core/src/main/java/com/dianping/cat/message/transport/TcpSocketTransport.java

-package com.dianping.cat.message.transport;
-
-import com.dianping.cat.message.Message;
-import com.dianping.cat.message.handler.MessageHandler;
-
-public class TcpSocketTransport implements Transport {
-	@Override
-   public void send(Message message) {
-   }
-
-	@Override
-	public void onMessage(MessageHandler handler) {
-	}
-
-	@Override
-   public void shutdown() {
-   }
-}

cat-core/src/main/java/com/dianping/cat/message/transport/UdpMulticastTransport.java

-package com.dianping.cat.message.transport;
-
-import com.dianping.cat.message.Message;
-import com.dianping.cat.message.handler.MessageHandler;
-
-public class UdpMulticastTransport implements Transport {
-	@Override
-	public void send(Message message) {
-	}
-
-	@Override
-	public void onMessage(MessageHandler handler) {
-	}
-
-	@Override
-	public void shutdown() {
-	}
-}

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

 <plexus>
 	<components>
 		<component>
-			<role>com.dianping.cat.transport.Transport</role>
+			<role>com.dianping.cat.message.io.Transport</role>
+			<role-hint>tcp-socket</role-hint>
+			<implementation>com.dianping.cat.message.io.TcpSocketTransport</implementation>
+		</component>
+		<component>
+			<role>com.dianping.cat.message.io.Transport</role>
+			<role-hint>udp-multicast</role-hint>
+			<implementation>com.dianping.cat.message.io.UdpMulticastTransport</implementation>
+		</component>
+		<component>
+			<role>com.dianping.cat.message.io.InMemoryQueue</role>
+			<implementation>com.dianping.cat.message.io.InMemoryQueue</implementation>
+		</component>
+		<component>
+			<role>com.dianping.cat.message.io.MessageSender</role>
 			<role-hint>in-memory</role-hint>
-			<implementation>com.dianping.cat.transport.InMemoryTransport</implementation>
+			<implementation>com.dianping.cat.message.io.InMemorySender</implementation>
+			<requirements>
+				<requirement>
+					<role>com.dianping.cat.message.io.InMemoryQueue</role>
+				</requirement>
+			</requirements>
 		</component>
 		<component>
-			<role>com.dianping.cat.transport.Transport</role>
-			<role-hint>tcp-socket</role-hint>
-			<implementation>com.dianping.cat.transport.TcpSocketTransport</implementation>
+			<role>com.dianping.cat.message.io.MessageReceiver</role>
+			<role-hint>in-memory</role-hint>
+			<implementation>com.dianping.cat.message.io.InMemoryReceiver</implementation>
+			<requirements>
+				<requirement>
+					<role>com.dianping.cat.message.io.InMemoryQueue</role>
+				</requirement>
+			</requirements>
 		</component>
 		<component>
-			<role>com.dianping.cat.transport.Transport</role>
-			<role-hint>udp-multicast</role-hint>
-			<implementation>com.dianping.cat.transport.UdpMulticastTransport</implementation>
+			<role>com.dianping.cat.message.handler.MessageHandler</role>
+			<implementation>com.dianping.cat.message.handler.MessageDispatcher</implementation>
+			<requirements>
+				<requirement>
+					<role>com.dianping.cat.message.io.MessageReceiver</role>
+					<role-hint>in-memory</role-hint>
+				</requirement>
+			</requirements>
 		</component>
 		<component>
-			<role>com.dianping.cat.transport.TransportManager</role>
-			<implementation>com.dianping.cat.transport.TransportManager</implementation>
+			<role>com.dianping.cat.message.broker.MessageBroker</role>
+			<role-hint>in-memory</role-hint>
+			<implementation>com.dianping.cat.message.broker.DefaultMessageBroker</implementation>
 			<requirements>
 				<requirement>
-					<role>com.dianping.cat.transport.Transport</role>
-					<role-hint>im-memory</role-hint>
+					<role>com.dianping.cat.message.io.MessageSender</role>
+					<role-hint>in-memory</role-hint>
+				</requirement>
+				<requirement>
+					<role>com.dianping.cat.message.io.MessageReceiver</role>
+					<role-hint>in-memory</role-hint>
 				</requirement>
 			</requirements>
 		</component>

cat-core/src/test/java/com/dianping/cat/message/EventTest.java

@@ -8,7 +8,7 @@ import org.junit.Test;
 import com.dianping.cat.Cat;
 
 public class EventTest {
-	public static MessageFactory CAT = Cat.getFactory();
+	public static MessageProducer CAT = Cat.getProducer();
 
 	@Test
 	public void testNormal() {

cat-core/src/test/java/com/dianping/cat/message/HeartbeatTest.java

@@ -5,7 +5,7 @@ import org.junit.Test;
 import com.dianping.cat.Cat;
 
 public class HeartbeatTest {
-	public static MessageFactory CAT = Cat.getFactory();
+	public static MessageProducer CAT = Cat.getProducer();
 
 	@Test
 	public void testStatus() {

cat-core/src/test/java/com/dianping/cat/message/TransactionTest.java

@@ -5,7 +5,7 @@ import org.junit.Test;
 import com.dianping.cat.Cat;
 
 public class TransactionTest {
-	public static MessageFactory CAT = Cat.getFactory();
+	public static MessageProducer CAT = Cat.getProducer();
 
 	@Test
 	public void testNormal() {

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

-package com.dianping.cat.message.transport;
+package com.dianping.cat.message.io;
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
@@ -10,45 +9,49 @@ import java.util.concurrent.Future;
 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.handler.MessageHandler;
+import com.site.lookup.ComponentTestCase;
 
-public class InMemoryTransportTest {
+@RunWith(JUnit4.class)
+public class InMemoryTest extends ComponentTestCase {
 	@Test
-	public void test() throws InterruptedException, ExecutionException {
-		final Transport transport = new InMemoryTransport();
+	public void test() throws Exception {
+		final MessageSender sender = lookup(MessageSender.class, "in-memory");
+		final MessageReceiver receiver = lookup(MessageReceiver.class, "in-memory");
 		final int len = 1000;
-		final StringBuilder sb = new StringBuilder(len * 4);
+		final StringBuilder sb = new StringBuilder(len);
 		ExecutorService pool = Executors.newFixedThreadPool(3);
 		List<Future<?>> futures = new ArrayList<Future<?>>();
 
 		futures.add(pool.submit(new Runnable() {
 			@Override
 			public void run() {
-				transport.onMessage(new MockMessageHandler(sb));
+				receiver.onMessage(new MockMessageHandler(sb));
 			}
 		}));
 		futures.add(pool.submit(new Runnable() {
 			@Override
 			public void run() {
 				for (int i = 0; i < len; i++) {
-					MockMessage m = new MockMessage();
-
-					m.setName(String.valueOf(i));
-					transport.send(m);
+					sender.send(new MockMessage());
 				}
 
-				transport.shutdown();
+				sender.shutdown();
+				receiver.shutdown();
 			}
 		}));
 
 		for (Future<?> future : futures) {
 			future.get();
 		}
+
 		pool.shutdown();
 
-		Assert.assertEquals(2890, sb.length());
+		Assert.assertEquals(len, sb.length());
 	}
 
 	static class MockMessage implements Message {
@@ -94,9 +97,9 @@ public class InMemoryTransportTest {
 		public void setStatus(String status) {
 		}
 
-      @Override
-      public void setStatus(Throwable e) {
-      }
+		@Override
+		public void setStatus(Throwable e) {
+		}
 	}
 
 	static class MockMessageHandler implements MessageHandler {
@@ -108,7 +111,7 @@ public class InMemoryTransportTest {
 
 		@Override
 		public void handle(Message message) {
-			m_sb.append(message.getName());
+			m_sb.append('.');
 		}
 	}
 }

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

-package com.dianping.cat.message.transport;
+package com.dianping.cat.message.io;
 
 import junit.framework.Assert;
 
@@ -6,6 +6,8 @@ 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;
 
 public class TransportManagerTest {
    @Test

pom.xml

@@ -16,9 +16,14 @@
    <dependencyManagement>
       <dependencies>
          <dependency>
-            <groupId>org.codehaus.plexus</groupId>
-            <artifactId>plexus-container-default</artifactId>
-            <version>1.0-alpha-47</version>
+            <groupId>com.site.common</groupId>
+            <artifactId>lookup</artifactId>
+            <version>1.0.0</version>
+         </dependency>
+         <dependency>
+            <groupId>com.site.common</groupId>
+            <artifactId>test-framework</artifactId>
+            <version>1.0.0</version>
          </dependency>
          <dependency>
             <groupId>junit</groupId>
@@ -26,14 +31,14 @@
             <version>4.8.1</version>
          </dependency>
          <dependency>
-            <groupId>com.site.common</groupId>
-            <artifactId>lookup</artifactId>
-            <version>1.0.0</version>
+            <groupId>org.codehaus.plexus</groupId>
+            <artifactId>plexus-container-default</artifactId>
+            <version>1.0-alpha-47</version>
          </dependency>
          <dependency>
-            <groupId>com.site.common</groupId>
-            <artifactId>test-framework</artifactId>
-            <version>1.0.0</version>
+            <groupId>org.jboss.netty</groupId>
+            <artifactId>netty</artifactId>
+            <version>3.2.7.Final</version>
          </dependency>
       </dependencies>
    </dependencyManagement>

settings.xml

 <?xml version="1.0" encoding="UTF-8"?>
 <settings xmlns="http://maven.apache.org/SETTINGS/1.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="http://maven.apache.org/SETTINGS/1.0.0 http://maven.apache.org/xsd/settings-1.0.0.xsd">
-   <localRepository>/var/mvnrepo</localRepository>
+   <localRepository>~/.m2/repository</localRepository>
    <servers>
       <server>
          <id>thirdparty</id>
@@ -92,6 +92,13 @@
                   <enabled>false</enabled>
                </snapshots>
             </repository>
+            <repository>
+               <id>repository.jboss.org</id>
+               <url>https://repository.jboss.org/nexus/content/repositories/releases/</url>
+               <snapshots>
+                  <enabled>false</enabled>
+               </snapshots>
+            </repository>
          </repositories>
          <pluginRepositories>
             <pluginRepository>