/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.catalina.tribes.demos;
import java.util.ArrayList;
import java.util.Properties;
import org.apache.catalina.tribes.Channel;
import org.apache.catalina.tribes.ManagedChannel;
import org.apache.catalina.tribes.Member;
import org.apache.catalina.tribes.group.GroupChannel;
import org.apache.catalina.tribes.group.interceptors.DomainFilterInterceptor;
import org.apache.catalina.tribes.group.interceptors.FragmentationInterceptor;
import org.apache.catalina.tribes.group.interceptors.GzipInterceptor;
import org.apache.catalina.tribes.group.interceptors.MessageDispatchInterceptor;
import org.apache.catalina.tribes.group.interceptors.OrderInterceptor;
import org.apache.catalina.tribes.group.interceptors.StaticMembershipInterceptor;
import org.apache.catalina.tribes.group.interceptors.TcpFailureDetector;
import org.apache.catalina.tribes.group.interceptors.ThroughputInterceptor;
import org.apache.catalina.tribes.membership.McastService;
import org.apache.catalina.tribes.membership.MemberImpl;
import org.apache.catalina.tribes.transport.Constants;
import org.apache.catalina.tribes.transport.MultiPointSender;
import org.apache.catalina.tribes.transport.ReceiverBase;
import org.apache.catalina.tribes.transport.ReplicationTransmitter;
public class ChannelCreator {
public static StringBuilder usage() {
StringBuilder buf = new StringBuilder();
buf.append(" [-bind tcpbindaddress]").append(" [-tcpselto tcpselectortimeout]")
.append(" [-tcpthreads tcpthreadcount]").append(" [-port tcplistenport]")
.append(" [-autobind tcpbindtryrange]").append(" [-ackto acktimeout]")
.append(" [-receiver org.apache.catalina.tribes.transport.nio.NioReceiver|org.apache.catalina.tribes.transport.bio.BioReceiver|]")
.append(" [-transport org.apache.catalina.tribes.transport.nio.PooledParallelSender|org.apache.catalina.tribes.transport.bio.PooledMultiSender]")
.append(" [-transport.xxx transport specific property]").append(" [-maddr multicastaddr]")
.append(" [-mport multicastport]").append(" [-mbind multicastbindaddr]")
.append(" [-mfreq multicastfrequency]").append(" [-mdrop multicastdroptime]")
.append(" [-gzip]")
.append(" [-static hostname:port (-static localhost:9999 -static 127.0.0.1:8888 can be repeated)]")
.append(" [-order]").append(" [-ordersize maxorderqueuesize]").append(" [-frag]")
.append(" [-fragsize maxmsgsize]").append(" [-throughput]").append(" [-failuredetect]")
.append(" [-async]").append(" [-asyncsize maxqueuesizeinkilobytes]");
return buf;
}
public static Channel createChannel(String[] args) throws Exception {
String bind = "auto";
int port = 4001;
String mbind = null;
boolean gzip = false;
int tcpseltimeout = 5000;
int tcpthreadcount = 4;
int acktimeout = 15000;
String mcastaddr = "228.0.0.5";
int mcastport = 45565;
long mcastfreq = 500;
long mcastdrop = 2000;
boolean order = false;
int ordersize = Integer.MAX_VALUE;
boolean frag = false;
int fragsize = 1024;
int autoBind = 10;
ArrayList<Member> staticMembers = new ArrayList<>();
Properties transportProperties = new Properties();
String transport = "org.apache.catalina.tribes.transport.nio.PooledParallelSender";
String receiver = "org.apache.catalina.tribes.transport.nio.NioReceiver";
boolean async = false;
int asyncsize = 1024 * 1024 * 50; // 50 MiB
boolean throughput = false;
boolean failuredetect = false;
for (int i = 0; i < args.length; i++) {
if ("-bind".equals(args[i])) {
bind = args[++i];
} else if ("-port".equals(args[i])) {
port = Integer.parseInt(args[++i]);
} else if ("-autobind".equals(args[i])) {
autoBind = Integer.parseInt(args[++i]);
} else if ("-tcpselto".equals(args[i])) {
tcpseltimeout = Integer.parseInt(args[++i]);
} else if ("-tcpthreads".equals(args[i])) {
tcpthreadcount = Integer.parseInt(args[++i]);
} else if ("-gzip".equals(args[i])) {
gzip = true;
} else if ("-async".equals(args[i])) {
async = true;
} else if ("-failuredetect".equals(args[i])) {
failuredetect = true;
} else if ("-asyncsize".equals(args[i])) {
asyncsize = Integer.parseInt(args[++i]);
System.out.println("Setting MessageDispatchInterceptor.maxQueueSize=" + asyncsize);
} else if ("-static".equals(args[i])) {
String d = args[++i];
String h = d.substring(0, d.indexOf(':'));
String p = d.substring(h.length() + 1);
Member m = new MemberImpl(h, Integer.parseInt(p), 2000);
staticMembers.add(m);
} else if ("-throughput".equals(args[i])) {
throughput = true;
} else if ("-order".equals(args[i])) {
order = true;
} else if ("-ordersize".equals(args[i])) {
ordersize = Integer.parseInt(args[++i]);
System.out.println("Setting OrderInterceptor.maxQueue=" + ordersize);
} else if ("-frag".equals(args[i])) {
frag = true;
} else if ("-fragsize".equals(args[i])) {
fragsize = Integer.parseInt(args[++i]);
System.out.println("Setting FragmentationInterceptor.maxSize=" + fragsize);
} else if ("-ackto".equals(args[i])) {
acktimeout = Integer.parseInt(args[++i]);
} else if ("-transport".equals(args[i])) {
transport = args[++i];
} else if (args[i] != null && args[i].startsWith("transport.")) {
String key = args[i];
String val = args[++i];
transportProperties.setProperty(key, val);
} else if ("-receiver".equals(args[i])) {
receiver = args[++i];
} else if ("-maddr".equals(args[i])) {
mcastaddr = args[++i];
} else if ("-mport".equals(args[i])) {
mcastport = Integer.parseInt(args[++i]);
} else if ("-mfreq".equals(args[i])) {
mcastfreq = Long.parseLong(args[++i]);
} else if ("-mdrop".equals(args[i])) {
mcastdrop = Long.parseLong(args[++i]);
} else if ("-mbind".equals(args[i])) {
mbind = args[++i];
}
}
System.out.println("Creating receiver class=" + receiver);
Class<?> cl = Class.forName(receiver, true, ChannelCreator.class.getClassLoader());
ReceiverBase rx = (ReceiverBase) cl.getConstructor().newInstance();
rx.setAddress(bind);
rx.setPort(port);
rx.setSelectorTimeout(tcpseltimeout);
rx.setMaxThreads(tcpthreadcount);
rx.setMinThreads(tcpthreadcount);
rx.getBind();
rx.setRxBufSize(Constants.DEFAULT_CLUSTER_MSG_BUFFER_SIZE);
rx.setTxBufSize(Constants.DEFAULT_CLUSTER_ACK_BUFFER_SIZE);
rx.setAutoBind(autoBind);
ReplicationTransmitter ps = new ReplicationTransmitter();
System.out.println("Creating transport class=" + transport);
MultiPointSender sender = (MultiPointSender) Class
.forName(transport, true, ChannelCreator.class.getClassLoader()).getConstructor().newInstance();
sender.setTimeout(acktimeout);
sender.setMaxRetryAttempts(2);
sender.setRxBufSize(Constants.DEFAULT_CLUSTER_MSG_BUFFER_SIZE);
sender.setTxBufSize(Constants.DEFAULT_CLUSTER_ACK_BUFFER_SIZE);
for (Object o : transportProperties.keySet()) {
String key = (String) o;
IntrospectionUtils.setProperty(sender, key, transportProperties.getProperty(key));
}
IntrospectionUtils.clear();
ps.setTransport(sender);
McastService service = new McastService();
service.setAddress(mcastaddr);
if (mbind != null) {
service.setMcastBindAddress(mbind);
}
service.setFrequency(mcastfreq);
service.setMcastDropTime(mcastdrop);
service.setPort(mcastport);
ManagedChannel channel = new GroupChannel();
channel.setChannelReceiver(rx);
channel.setChannelSender(ps);
channel.setMembershipService(service);
if (throughput) {
channel.addInterceptor(new ThroughputInterceptor());
}
if (gzip) {
channel.addInterceptor(new GzipInterceptor());
}
if (frag) {
FragmentationInterceptor fi = new FragmentationInterceptor();
fi.setMaxSize(fragsize);
channel.addInterceptor(fi);
}
if (order) {
OrderInterceptor oi = new OrderInterceptor();
oi.setMaxQueue(ordersize);
channel.addInterceptor(oi);
}
if (async) {
MessageDispatchInterceptor mi = new MessageDispatchInterceptor();
mi.setMaxQueueSize(asyncsize);
channel.addInterceptor(mi);
System.out.println("Added MessageDispatchInterceptor");
}
if (failuredetect) {
TcpFailureDetector tcpfi = new TcpFailureDetector();
channel.addInterceptor(tcpfi);
}
if (staticMembers.size() > 0) {
StaticMembershipInterceptor smi = new StaticMembershipInterceptor();
for (Member staticMember : staticMembers) {
smi.addStaticMember(staticMember);
}
channel.addInterceptor(smi);
}
byte[] domain = new byte[] { 1, 2, 3, 4, 5, 6, 7, 8, 9, 0 };
channel.getMembershipService().setDomain(domain);
DomainFilterInterceptor filter = new DomainFilterInterceptor();
filter.setDomain(domain);
channel.addInterceptor(filter);
return channel;
}
}