2 * Copyright 2005-2011 the original author or authors.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
16 package org.wamblee.xmlrouter.impl;
18 import java.util.ArrayList;
19 import java.util.Collection;
20 import java.util.Collections;
21 import java.util.HashSet;
22 import java.util.LinkedHashMap;
23 import java.util.List;
26 import java.util.concurrent.atomic.AtomicLong;
27 import java.util.logging.Level;
28 import java.util.logging.Logger;
30 import javax.xml.transform.dom.DOMSource;
32 import org.wamblee.general.Clock;
33 import org.wamblee.xml.XMLDocument;
34 import org.wamblee.xmlrouter.common.Id;
35 import org.wamblee.xmlrouter.config.DocumentType;
36 import org.wamblee.xmlrouter.config.Filter;
37 import org.wamblee.xmlrouter.config.RouterConfig;
38 import org.wamblee.xmlrouter.config.Transformation;
39 import org.wamblee.xmlrouter.listener.EventInfo;
40 import org.wamblee.xmlrouter.listener.EventListener;
41 import org.wamblee.xmlrouter.publish.Gateway;
42 import org.wamblee.xmlrouter.subscribe.Destination;
43 import org.wamblee.xmlrouter.subscribe.DestinationRegistry;
47 public class XMLRouter implements RouterConfig, Gateway, DestinationRegistry {
49 private static final Logger LOGGER = Logger.getLogger(XMLRouter.class
52 private EventListener listener;
54 private AtomicLong nextEventId;
55 private AtomicLong sequenceNumbers;
56 private Map<Id<DocumentType>, DocumentType> documentTypes;
57 private Transformations transformations;
58 private Map<Id<Filter>, Filter> filters;
59 private Map<Id<Destination>, Destination> destinations;
61 public XMLRouter(Clock aClock, EventListener aListener) {
64 nextEventId = new AtomicLong(clock.currentTimeMillis());
65 sequenceNumbers = new AtomicLong(1);
66 documentTypes = new LinkedHashMap<Id<DocumentType>, DocumentType>();
67 transformations = new Transformations();
68 filters = new LinkedHashMap<Id<Filter>, Filter>();
69 destinations = new LinkedHashMap<Id<Destination>, Destination>();
73 public Id<DocumentType> addDocumentType(DocumentType aType) {
74 long seqno = sequenceNumbers.getAndIncrement();
75 documentTypes.put(new Id<DocumentType>(seqno), aType);
76 return new Id<DocumentType>(seqno);
80 public void removeDocumentType(Id<DocumentType> aId) {
81 documentTypes.remove(aId);
85 public Collection<Id<DocumentType>> getDocumentTypes() {
86 return Collections.unmodifiableCollection(documentTypes.keySet());
90 public DocumentType getDocumentType(Id<DocumentType> aId) {
91 return documentTypes.get(aId);
95 public Id<Transformation> addTransformation(Transformation aTransformation) {
96 return transformations.addTransformation(aTransformation);
100 public void removeTransformation(Id<Transformation> aId) {
101 transformations.removeTransformation(aId);
105 public Collection<Id<Transformation>> getTransformations() {
106 return transformations.getTransformations();
110 public Transformation getTransformation(Id<Transformation> aId) {
111 return transformations.getTransformation(aId);
115 public Id<Filter> addFilter(Filter aFilter) {
116 long seqno = sequenceNumbers.getAndIncrement();
117 filters.put(new Id<Filter>(seqno), aFilter);
118 return new Id<Filter>(seqno);
122 public void removeFilter(Id<Filter> aId) {
127 public Collection<Id<Filter>> getFilters() {
128 return Collections.unmodifiableCollection(filters.keySet());
132 public Filter getFilter(Id<Filter> aId) {
133 return filters.get(aId);
137 public void publish(String aSource, DOMSource aEvent) {
139 long time = clock.currentTimeMillis();
140 Id<DOMSource> id = new Id<DOMSource>(nextEventId.getAndIncrement());
141 List<String> types = determineDocumentTypes(aEvent);
142 EventInfo info = new EventInfo(time, aSource, id, types, aEvent);
144 boolean delivered = false;
147 List<String> filteredInputTypes = determineFilteredInputTypes(
149 if (filteredInputTypes.isEmpty()) {
150 if (LOGGER.isLoggable(Level.FINE)) {
151 String doc = new XMLDocument(aEvent).print(true);
155 "Event ''0}'' from source {1} removed because of filters.",
156 new Object[] { doc, aSource });
160 // get the reachable target types through transformations.
162 // It is possible that a given event belongs to multiple input
164 // This is however certainly not the main case.
166 for (String inputType : filteredInputTypes) {
167 boolean result = deliverEvent(info, inputType);
168 delivered = delivered || result;
172 destinationNotFound(aSource, aEvent);
173 listener.notDelivered(info);
178 private boolean deliverEvent(EventInfo aInfo, String aInputType) {
180 boolean delivered = false;
181 Set<String> possibleTargetTypes = new HashSet<String>();
182 possibleTargetTypes.addAll(transformations
183 .getPossibleTargetTypes(aInputType));
185 // ask each destination what target types, if any they want to have.
186 for (Map.Entry<Id<Destination>, Destination> entry : destinations
188 Id<Destination> destinationId = entry.getKey();
189 Destination destination = entry.getValue();
190 Collection<String> requested = destination
191 .chooseFromTargetTypes(possibleTargetTypes);
192 if (!requested.isEmpty()) {
193 // Deliver to the destination.
194 for (String targetType : requested) {
195 TransformationPath path = transformations.getPath(
196 aInputType, targetType);
197 List<Transformation> ts = path.getTransformations();
199 boolean allowed = true;
200 DOMSource transformed = aInfo.getEvent();
201 while (i < ts.size() && allowed && transformed != null) {
202 Transformation t = ts.get(i);
203 DOMSource orig = transformed;
204 transformed = t.transform(transformed);
205 if (transformed == null) {
206 transformationReturnedNull(aInfo.getSource(),
207 aInfo.getEvent(), aInputType, t, orig);
210 if (!isAllowedByFilters(t.getToType(), transformed)) {
215 if (allowed && transformed != null) {
216 // all transformations done and all filters still
218 boolean result = destination.receive(transformed);
219 listener.delivered(aInfo, ts, destinationId.getId(),
220 destination.getName(), result);
221 delivered = delivered || result;
230 private List<String> determineFilteredInputTypes(List<String> aTypes,
233 // apply filters to the input
234 List<String> filteredTypes = new ArrayList<String>();
235 for (String type : aTypes) {
236 boolean allowed = isAllowedByFilters(type, aEvent);
238 filteredTypes.add(type);
241 return filteredTypes;
244 private boolean isAllowedByFilters(String aType, DOMSource aEvent) {
245 boolean allowed = true;
246 for (Filter filter : filters.values()) {
247 if (!filter.isAllowed(aType, aEvent)) {
254 private List<String> determineDocumentTypes(DOMSource aEvent) {
255 List<String> res = new ArrayList<String>();
256 for (DocumentType type : documentTypes.values()) {
257 if (type.isInstance(aEvent)) {
258 res.add(type.getName());
264 private void logEvent(String aMessage, String aSource, DOMSource aEvent,
265 Exception aException) {
266 LOGGER.log(Level.WARNING, aMessage + ": source '" + aSource +
267 "': Event: '" + new XMLDocument(aEvent).print(true) + "'",
271 private void logEvent(String aMessage, String aSource, DOMSource aEvent) {
272 LOGGER.log(Level.WARNING,
273 aMessage + ": " + eventToString(aSource, aEvent));
276 private String eventToString(String aSource, DOMSource aEvent) {
277 return "source '" + aSource + "': Event: '" +
278 new XMLDocument(aEvent).print(true) + "'";
281 private void transformationReturnedNull(String aSource, DOMSource aEvent,
282 String aInputType, Transformation aT, DOMSource aTransformed) {
283 LOGGER.log(Level.WARNING, "Transformation returned null for event " +
284 eventToString(aSource, aEvent) + " inputType '" + aInputType +
285 "', transformation '" + aT + "' document to transform " +
286 new XMLDocument(aTransformed).print(true));
289 private void destinationNotFound(String aSource, DOMSource aEvent) {
290 LOGGER.log(Level.WARNING, "No destination found for event: " +
291 eventToString(aSource, aEvent));
295 public Id<Destination> registerDestination(Destination aDestination) {
296 notNull("destination", aDestination);
297 long seqno = sequenceNumbers.getAndIncrement();
298 Id<Destination> id = new Id<Destination>(seqno);
299 destinations.put(id, new RobustDestination(id, aDestination));
304 public void unregisterDestination(Id<Destination> aId) {
305 destinations.remove(aId);
308 private void notNull(String aName, Object aValue) {
309 if (aValue == null) {
310 throw new IllegalArgumentException("Parameter '" + aName +
311 "' may not be null");