KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > org > apache > activemq > transport > udp > UdpTransportFactory


1 /**
2  *
3  * Licensed to the Apache Software Foundation (ASF) under one or more
4  * contributor license agreements. See the NOTICE file distributed with
5  * this work for additional information regarding copyright ownership.
6  * The ASF licenses this file to You under the Apache License, Version 2.0
7  * (the "License"); you may not use this file except in compliance with
8  * the License. You may obtain a copy of the License at
9  *
10  * http://www.apache.org/licenses/LICENSE-2.0
11  *
12  * Unless required by applicable law or agreed to in writing, software
13  * distributed under the License is distributed on an "AS IS" BASIS,
14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15  * See the License for the specific language governing permissions and
16  * limitations under the License.
17  */
18 package org.apache.activemq.transport.udp;
19
20 import java.io.IOException JavaDoc;
21 import java.net.URI JavaDoc;
22 import java.net.URISyntaxException JavaDoc;
23 import java.net.UnknownHostException JavaDoc;
24 import java.util.HashMap JavaDoc;
25 import java.util.Map JavaDoc;
26
27 import org.apache.activemq.openwire.OpenWireFormat;
28 import org.apache.activemq.transport.CommandJoiner;
29 import org.apache.activemq.transport.InactivityMonitor;
30 import org.apache.activemq.transport.Transport;
31 import org.apache.activemq.transport.TransportFactory;
32 import org.apache.activemq.transport.TransportLogger;
33 import org.apache.activemq.transport.TransportServer;
34 import org.apache.activemq.transport.reliable.DefaultReplayStrategy;
35 import org.apache.activemq.transport.reliable.ExceptionIfDroppedReplayStrategy;
36 import org.apache.activemq.transport.reliable.ReliableTransport;
37 import org.apache.activemq.transport.reliable.ReplayStrategy;
38 import org.apache.activemq.transport.reliable.Replayer;
39 import org.apache.activemq.util.IOExceptionSupport;
40 import org.apache.activemq.util.IntrospectionSupport;
41 import org.apache.activemq.util.URISupport;
42 import org.apache.activemq.wireformat.WireFormat;
43
44 public class UdpTransportFactory extends TransportFactory {
45
46     public TransportServer doBind(String JavaDoc brokerId, final URI JavaDoc location) throws IOException JavaDoc {
47         try {
48             Map JavaDoc options = new HashMap JavaDoc(URISupport.parseParamters(location));
49             if (options.containsKey("port")) {
50                 throw new IllegalArgumentException JavaDoc("The port property cannot be specified on a UDP server transport - please use the port in the URI syntax");
51             }
52             WireFormat wf = createWireFormat(options);
53             int port = location.getPort();
54             OpenWireFormat openWireFormat = asOpenWireFormat(wf);
55             UdpTransport transport = new UdpTransport(openWireFormat, port);
56
57             Transport configuredTransport = configure(transport, wf, options, true);
58             UdpTransportServer server = new UdpTransportServer(location, transport, configuredTransport, createReplayStrategy());
59             return server;
60         }
61         catch (URISyntaxException JavaDoc e) {
62             throw IOExceptionSupport.create(e);
63         }
64         catch (Exception JavaDoc e) {
65             throw IOExceptionSupport.create(e);
66         }
67     }
68
69     public Transport configure(Transport transport, WireFormat format, Map JavaDoc options) throws Exception JavaDoc {
70         return configure(transport, format, options, false);
71     }
72
73     public Transport compositeConfigure(Transport transport, WireFormat format, Map JavaDoc options) {
74         IntrospectionSupport.setProperties(transport, options);
75         final UdpTransport udpTransport = (UdpTransport) transport;
76
77         // deal with fragmentation
78 transport = new CommandJoiner(transport, asOpenWireFormat(format));
79
80         if (udpTransport.isTrace()) {
81             transport = new TransportLogger(transport);
82         }
83
84         transport = new InactivityMonitor(transport);
85
86         if (format instanceof OpenWireFormat) {
87             transport = configureClientSideNegotiator(transport, format, udpTransport);
88         }
89
90         return transport;
91     }
92
93     protected Transport createTransport(URI JavaDoc location, WireFormat wf) throws UnknownHostException JavaDoc, IOException JavaDoc {
94         OpenWireFormat wireFormat = asOpenWireFormat(wf);
95         return new UdpTransport(wireFormat, location);
96     }
97
98     /**
99      * Configures the transport
100      *
101      * @param acceptServer
102      * true if this transport is used purely as an 'accept' transport
103      * for new connections which work like TCP SocketServers where
104      * new connections spin up a new separate UDP transport
105      */
106     protected Transport configure(Transport transport, WireFormat format, Map JavaDoc options, boolean acceptServer) throws Exception JavaDoc {
107         IntrospectionSupport.setProperties(transport, options);
108         UdpTransport udpTransport = (UdpTransport) transport;
109
110         OpenWireFormat openWireFormat = asOpenWireFormat(format);
111
112         if (udpTransport.isTrace()) {
113             transport = new TransportLogger(transport);
114         }
115
116         transport = new InactivityMonitor(transport);
117
118         if (!acceptServer && format instanceof OpenWireFormat) {
119             transport = configureClientSideNegotiator(transport, format, udpTransport);
120         }
121
122         // deal with fragmentation
123
124         if (acceptServer) {
125             // lets not support a buffer of messages to enable reliable
126 // messaging on the 'accept server' transport
127 udpTransport.setReplayEnabled(false);
128
129             // we don't want to do reliable checks on this transport as we
130 // delegate to one that does
131 transport = new CommandJoiner(transport, openWireFormat);
132             return transport;
133         }
134         else {
135             ReliableTransport reliableTransport = new ReliableTransport(transport, udpTransport);
136             Replayer replayer = reliableTransport.getReplayer();
137             reliableTransport.setReplayStrategy(createReplayStrategy(replayer));
138
139             // Joiner must be on outside as the inbound messages must be
140 // processed by the reliable transport first
141 return new CommandJoiner(reliableTransport, openWireFormat);
142         }
143     }
144
145     protected ReplayStrategy createReplayStrategy(Replayer replayer) {
146         if (replayer != null) {
147             return new DefaultReplayStrategy(5);
148         }
149         return new ExceptionIfDroppedReplayStrategy(1);
150     }
151
152     protected ReplayStrategy createReplayStrategy() {
153         return new DefaultReplayStrategy(5);
154     }
155
156     protected Transport configureClientSideNegotiator(Transport transport, WireFormat format, final UdpTransport udpTransport) {
157         return new ResponseRedirectInterceptor(transport, udpTransport);
158     }
159
160     protected OpenWireFormat asOpenWireFormat(WireFormat wf) {
161         OpenWireFormat answer = (OpenWireFormat) wf;
162         return answer;
163     }
164 }
165
Popular Tags