001: /*
002: * Licensed to the Apache Software Foundation (ASF) under one or more
003: * contributor license agreements. See the NOTICE file distributed with
004: * this work for additional information regarding copyright ownership.
005: * The ASF licenses this file to You under the Apache License, Version 2.0
006: * (the "License"); you may not use this file except in compliance with
007: * the License. You may obtain a copy of the License at
008: *
009: * http://www.apache.org/licenses/LICENSE-2.0
010: *
011: * Unless required by applicable law or agreed to in writing, software
012: * distributed under the License is distributed on an "AS IS" BASIS,
013: * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014: * See the License for the specific language governing permissions and
015: * limitations under the License.
016: */
017: package org.apache.catalina.tribes.demos;
018:
019: import java.util.Iterator;
020: import java.util.Properties;
021:
022: import org.apache.catalina.tribes.Channel;
023: import org.apache.catalina.tribes.ManagedChannel;
024: import org.apache.catalina.tribes.group.GroupChannel;
025: import org.apache.catalina.tribes.group.interceptors.FragmentationInterceptor;
026: import org.apache.catalina.tribes.group.interceptors.GzipInterceptor;
027: import org.apache.catalina.tribes.group.interceptors.MessageDispatchInterceptor;
028: import org.apache.catalina.tribes.group.interceptors.OrderInterceptor;
029: import org.apache.catalina.tribes.membership.McastService;
030: import org.apache.catalina.tribes.transport.MultiPointSender;
031: import org.apache.catalina.tribes.transport.ReceiverBase;
032: import org.apache.catalina.tribes.transport.ReplicationTransmitter;
033: import org.apache.catalina.tribes.group.interceptors.ThroughputInterceptor;
034: import org.apache.catalina.tribes.group.interceptors.MessageDispatch15Interceptor;
035: import org.apache.catalina.tribes.group.interceptors.TcpFailureDetector;
036: import org.apache.catalina.tribes.group.interceptors.DomainFilterInterceptor;
037: import java.util.ArrayList;
038: import org.apache.catalina.tribes.membership.MemberImpl;
039: import org.apache.catalina.tribes.group.interceptors.StaticMembershipInterceptor;
040: import org.apache.catalina.tribes.Member;
041:
042: /**
043: * <p>Title: </p>
044: *
045: * <p>Description: </p>
046: *
047: *
048: * <p>Company: </p>
049: *
050: * @author fhanik
051: * @version 1.0
052: */
053: public class ChannelCreator {
054:
055: public static StringBuffer usage() {
056: StringBuffer buf = new StringBuffer();
057: buf
058: .append("\n\t\t[-bind tcpbindaddress]")
059: .append("\n\t\t[-tcpselto tcpselectortimeout]")
060: .append("\n\t\t[-tcpthreads tcpthreadcount]")
061: .append("\n\t\t[-port tcplistenport]")
062: .append("\n\t\t[-autobind tcpbindtryrange]")
063: .append("\n\t\t[-ackto acktimeout]")
064: .append(
065: "\n\t\t[-receiver org.apache.catalina.tribes.transport.nio.NioReceiver|org.apache.catalina.tribes.transport.bio.BioReceiver|]")
066: .append(
067: "\n\t\t[-transport org.apache.catalina.tribes.transport.nio.PooledParallelSender|org.apache.catalina.tribes.transport.bio.PooledMultiSender]")
068: .append(
069: "\n\t\t[-transport.xxx transport specific property]")
070: .append("\n\t\t[-maddr multicastaddr]")
071: .append("\n\t\t[-mport multicastport]")
072: .append("\n\t\t[-mbind multicastbindaddr]")
073: .append("\n\t\t[-mfreq multicastfrequency]")
074: .append("\n\t\t[-mdrop multicastdroptime]")
075: .append("\n\t\t[-gzip]")
076: .append(
077: "\n\t\t[-static hostname:port (-static localhost:9999 -static 127.0.0.1:8888 can be repeated)]")
078: .append("\n\t\t[-order]").append(
079: "\n\t\t[-ordersize maxorderqueuesize]").append(
080: "\n\t\t[-frag]").append(
081: "\n\t\t[-fragsize maxmsgsize]").append(
082: "\n\t\t[-throughput]").append(
083: "\n\t\t[-failuredetect]").append(
084: "\n\t\t[-async]").append(
085: "\n\t\t[-asyncsize maxqueuesizeinkilobytes]");
086: return buf;
087:
088: }
089:
090: public static Channel createChannel(String[] args) throws Exception {
091: String bind = "auto";
092: int port = 4001;
093: String mbind = null;
094: boolean gzip = false;
095: int tcpseltimeout = 5000;
096: int tcpthreadcount = 4;
097: int acktimeout = 15000;
098: String mcastaddr = "228.0.0.5";
099: int mcastport = 45565;
100: long mcastfreq = 500;
101: long mcastdrop = 2000;
102: boolean order = false;
103: int ordersize = Integer.MAX_VALUE;
104: boolean frag = false;
105: int fragsize = 1024;
106: int autoBind = 10;
107: ArrayList staticMembers = new ArrayList();
108: Properties transportProperties = new Properties();
109: String transport = "org.apache.catalina.tribes.transport.nio.PooledParallelSender";
110: String receiver = "org.apache.catalina.tribes.transport.nio.NioReceiver";
111: boolean async = false;
112: int asyncsize = 1024 * 1024 * 50; //50MB
113: boolean throughput = false;
114: boolean failuredetect = false;
115:
116: for (int i = 0; i < args.length; i++) {
117: if ("-bind".equals(args[i])) {
118: bind = args[++i];
119: } else if ("-port".equals(args[i])) {
120: port = Integer.parseInt(args[++i]);
121: } else if ("-autobind".equals(args[i])) {
122: autoBind = Integer.parseInt(args[++i]);
123: } else if ("-tcpselto".equals(args[i])) {
124: tcpseltimeout = Integer.parseInt(args[++i]);
125: } else if ("-tcpthreads".equals(args[i])) {
126: tcpthreadcount = Integer.parseInt(args[++i]);
127: } else if ("-gzip".equals(args[i])) {
128: gzip = true;
129: } else if ("-async".equals(args[i])) {
130: async = true;
131: } else if ("-failuredetect".equals(args[i])) {
132: failuredetect = true;
133: } else if ("-asyncsize".equals(args[i])) {
134: asyncsize = Integer.parseInt(args[++i]);
135: System.out
136: .println("Setting MessageDispatchInterceptor.maxQueueSize="
137: + asyncsize);
138: } else if ("-static".equals(args[i])) {
139: String d = args[++i];
140: String h = d.substring(0, d.indexOf(":"));
141: String p = d.substring(h.length() + 1);
142: MemberImpl m = new MemberImpl(h, Integer.parseInt(p),
143: 2000);
144: staticMembers.add(m);
145: } else if ("-throughput".equals(args[i])) {
146: throughput = true;
147: } else if ("-order".equals(args[i])) {
148: order = true;
149: } else if ("-ordersize".equals(args[i])) {
150: ordersize = Integer.parseInt(args[++i]);
151: System.out.println("Setting OrderInterceptor.maxQueue="
152: + ordersize);
153: } else if ("-frag".equals(args[i])) {
154: frag = true;
155: } else if ("-fragsize".equals(args[i])) {
156: fragsize = Integer.parseInt(args[++i]);
157: System.out
158: .println("Setting FragmentationInterceptor.maxSize="
159: + fragsize);
160: } else if ("-ackto".equals(args[i])) {
161: acktimeout = Integer.parseInt(args[++i]);
162: } else if ("-transport".equals(args[i])) {
163: transport = args[++i];
164: } else if (args[i] != null
165: && args[i].startsWith("transport.")) {
166: String key = args[i];
167: String val = args[++i];
168: transportProperties.setProperty(key, val);
169: } else if ("-receiver".equals(args[i])) {
170: receiver = args[++i];
171: } else if ("-maddr".equals(args[i])) {
172: mcastaddr = args[++i];
173: } else if ("-mport".equals(args[i])) {
174: mcastport = Integer.parseInt(args[++i]);
175: } else if ("-mfreq".equals(args[i])) {
176: mcastfreq = Long.parseLong(args[++i]);
177: } else if ("-mdrop".equals(args[i])) {
178: mcastdrop = Long.parseLong(args[++i]);
179: } else if ("-mbind".equals(args[i])) {
180: mbind = args[++i];
181: }
182: }
183:
184: System.out.println("Creating receiver class=" + receiver);
185: Class cl = Class.forName(receiver, true, ChannelCreator.class
186: .getClassLoader());
187: ReceiverBase rx = (ReceiverBase) cl.newInstance();
188: rx.setTcpListenAddress(bind);
189: rx.setTcpListenPort(port);
190: rx.setTcpSelectorTimeout(tcpseltimeout);
191: rx.setTcpThreadCount(tcpthreadcount);
192: rx.getBind();
193: rx.setRxBufSize(43800);
194: rx.setTxBufSize(25188);
195: rx.setAutoBind(autoBind);
196:
197: ReplicationTransmitter ps = new ReplicationTransmitter();
198: System.out.println("Creating transport class=" + transport);
199: MultiPointSender sender = (MultiPointSender) Class.forName(
200: transport, true, ChannelCreator.class.getClassLoader())
201: .newInstance();
202: sender.setTimeout(acktimeout);
203: sender.setMaxRetryAttempts(2);
204: sender.setRxBufSize(43800);
205: sender.setTxBufSize(25188);
206:
207: Iterator i = transportProperties.keySet().iterator();
208: while (i.hasNext()) {
209: String key = (String) i.next();
210: IntrospectionUtils.setProperty(sender, key,
211: transportProperties.getProperty(key));
212: }
213: ps.setTransport(sender);
214:
215: McastService service = new McastService();
216: service.setMcastAddr(mcastaddr);
217: if (mbind != null)
218: service.setMcastBindAddress(mbind);
219: service.setMcastFrequency(mcastfreq);
220: service.setMcastDropTime(mcastdrop);
221: service.setMcastPort(mcastport);
222:
223: ManagedChannel channel = new GroupChannel();
224: channel.setChannelReceiver(rx);
225: channel.setChannelSender(ps);
226: channel.setMembershipService(service);
227:
228: if (throughput)
229: channel.addInterceptor(new ThroughputInterceptor());
230: if (gzip)
231: channel.addInterceptor(new GzipInterceptor());
232: if (frag) {
233: FragmentationInterceptor fi = new FragmentationInterceptor();
234: fi.setMaxSize(fragsize);
235: channel.addInterceptor(fi);
236: }
237: if (order) {
238: OrderInterceptor oi = new OrderInterceptor();
239: oi.setMaxQueue(ordersize);
240: channel.addInterceptor(oi);
241: }
242:
243: if (async) {
244: MessageDispatchInterceptor mi = new MessageDispatch15Interceptor();
245: mi.setMaxQueueSize(asyncsize);
246: channel.addInterceptor(mi);
247: System.out.println("Added MessageDispatchInterceptor");
248: }
249:
250: if (failuredetect) {
251: TcpFailureDetector tcpfi = new TcpFailureDetector();
252: channel.addInterceptor(tcpfi);
253: }
254: if (staticMembers.size() > 0) {
255: StaticMembershipInterceptor smi = new StaticMembershipInterceptor();
256: for (int x = 0; x < staticMembers.size(); x++) {
257: smi.addStaticMember((Member) staticMembers.get(x));
258: }
259: channel.addInterceptor(smi);
260: }
261:
262: byte[] domain = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 0 };
263: ((McastService) channel.getMembershipService())
264: .setDomain(domain);
265: DomainFilterInterceptor filter = new DomainFilterInterceptor();
266: filter.setDomain(domain);
267: channel.addInterceptor(filter);
268: return channel;
269: }
270:
271: }
|