001: /*
002: * <copyright>
003: *
004: * Copyright 1997-2004 BBNT Solutions, LLC
005: * under sponsorship of the Defense Advanced Research Projects
006: * Agency (DARPA).
007: *
008: * You can redistribute this software and/or modify it under the
009: * terms of the Cougaar Open Source License as published on the
010: * Cougaar Open Source Website (www.cougaar.org).
011: *
012: * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
013: * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
014: * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
015: * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
016: * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
017: * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
018: * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
019: * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
020: * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
021: * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
022: * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
023: *
024: * </copyright>
025: */
026:
027: package org.cougaar.mts.std;
028:
029: import java.util.ArrayList;
030: import java.util.Iterator;
031:
032: import org.cougaar.core.mts.Message;
033: import org.cougaar.core.mts.MessageAddress;
034: import org.cougaar.core.mts.MessageAttributes;
035: import org.cougaar.core.mts.MessageTransportWatcher;
036: import org.cougaar.core.service.MessageWatcherService;
037: import org.cougaar.mts.base.MessageDeliverer;
038: import org.cougaar.mts.base.MessageDelivererDelegateImplBase;
039: import org.cougaar.mts.base.MessageTransportServiceProvider;
040: import org.cougaar.mts.base.MisdeliveredMessageException;
041: import org.cougaar.mts.base.SendQueue;
042: import org.cougaar.mts.base.SendQueueDelegateImplBase;
043: import org.cougaar.mts.base.StandardAspect;
044:
045: /**
046: * This Aspect is used in conjunction with {@link
047: * MessageWatcherServiceImpl}, the implementaion of the {@link
048: * MessageWatcherService}. Both are instantiated by the {@link
049: * MessageTransportServiceProvider}, which is also the provider of the
050: * {@link MessageWatcherService}. The actual "watching" happens in
051: * this Aspect. The service is a core front-end.
052: */
053: public class WatcherAspect extends StandardAspect {
054: private ArrayList watchers;
055:
056: public WatcherAspect() {
057: this .watchers = new ArrayList();
058: }
059:
060: public Object getDelegate(Object delegate, Class type) {
061: if (type == SendQueue.class) {
062: return new SendQueueDelegate((SendQueue) delegate);
063: } else if (type == MessageDeliverer.class) {
064: return new MessageDelivererDelegate(
065: (MessageDeliverer) delegate);
066: } else {
067: return null;
068: }
069: }
070:
071: synchronized void addWatcher(MessageTransportWatcher watcher) {
072: watchers.add(watcher);
073: }
074:
075: synchronized void removeWatcher(MessageTransportWatcher watcher) {
076: watchers.remove(watcher);
077: }
078:
079: // Should the watchers see the AttributedMessage or its contents?
080: private void notifyWatchersOfSend(AttributedMessage message) {
081: Message rawMessage = message.getRawMessage();
082: Iterator itr = watchers.iterator();
083: synchronized (this ) {
084: while (itr.hasNext()) {
085: MessageTransportWatcher w = (MessageTransportWatcher) itr
086: .next();
087: if (loggingService.isDebugEnabled()) {
088: loggingService.debug("Notifying " + w + " of send");
089: }
090: w.messageSent(rawMessage);
091: }
092: }
093: }
094:
095: private void notifyWatchersOfReceive(AttributedMessage message) {
096: Message rawMessage = message.getRawMessage();
097: Iterator itr = watchers.iterator();
098: synchronized (this ) {
099: while (itr.hasNext()) {
100: MessageTransportWatcher w = (MessageTransportWatcher) itr
101: .next();
102: if (loggingService.isDebugEnabled()) {
103: loggingService.debug("Notifying " + w
104: + " of receive");
105: }
106: w.messageReceived(rawMessage);
107: }
108: }
109: }
110:
111: public class SendQueueDelegate extends SendQueueDelegateImplBase {
112: public SendQueueDelegate(SendQueue queue) {
113: super (queue);
114: }
115:
116: public void sendMessage(AttributedMessage message) {
117: super .sendMessage(message);
118: notifyWatchersOfSend(message);
119: }
120:
121: }
122:
123: public class MessageDelivererDelegate extends
124: MessageDelivererDelegateImplBase {
125: public MessageDelivererDelegate(MessageDeliverer deliverer) {
126: super (deliverer);
127: }
128:
129: public MessageAttributes deliverMessage(
130: AttributedMessage message, MessageAddress dest)
131: throws MisdeliveredMessageException {
132: MessageAttributes result = super.deliverMessage(message,
133: dest);
134: notifyWatchersOfReceive(message);
135: return result;
136: }
137:
138: }
139: }
|