/*
* 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.log4j.receivers.net;
import org.apache.log4j.component.plugins.Pauseable;
import org.apache.log4j.component.plugins.Receiver;
import org.apache.log4j.net.ZeroConfSupport;
import org.apache.log4j.receivers.spi.Decoder;
import org.apache.log4j.spi.LoggingEvent;
import java.io.IOException;
import java.net.*;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
/**
* Multicast-based receiver. Accepts LoggingEvents encoded using
* MulticastAppender and XMLLayout. The the XML data is converted
* back to a LoggingEvent and is posted.
*
* @author Scott Deboy <sdeboy@apache.org>
*
*/
public class MulticastReceiver extends Receiver implements PortBased,
AddressBased, Pauseable {
private static final int PACKET_LENGTH = 16384;
private int port;
private String address;
private String encoding;
private MulticastSocket socket = null;
//default to log4j xml decoder
private String decoder = "org.apache.log4j.xml.XMLDecoder";
private Decoder decoderImpl;
private MulticastHandlerThread handlerThread;
private MulticastReceiverThread receiverThread;
private boolean paused;
private boolean advertiseViaMulticastDNS;
private ZeroConfSupport zeroConf;
/**
* The MulticastDNS zone advertised by a MulticastReceiver
*/
public static final String ZONE = "_log4j_xml_mcast_receiver.local.";
public String getDecoder() {
return decoder;
}
public void setDecoder(String decoder) {
this.decoder = decoder;
}
public int getPort() {
return port;
}
public void setPort(int port) {
this.port = port;
}
public String getAddress() {
return address;
}
/**
The <b>Encoding</b> option specifies how the bytes are encoded. If this option is not specified,
the system encoding will be used.
*/
public void setEncoding(String encoding) {
this.encoding = encoding;
}
/**
Returns value of the <b>Encoding</b> option.
*/
public String getEncoding() {
return encoding;
}
public synchronized void shutdown() {
active = false;
if (advertiseViaMulticastDNS) {
zeroConf.unadvertise();
}
if (handlerThread != null) {
handlerThread.interrupt();
}
if (receiverThread != null) {
receiverThread.interrupt();
}
if (socket != null) {
socket.close();
}
}
public void setAddress(String address) {
this.address = address;
}
public boolean isPaused() {
return paused;
}
public void setPaused(boolean b) {
paused = b;
}
public void activateOptions() {
InetAddress addr = null;
try {
Class c = Class.forName(decoder);
Object o = c.newInstance();
if (o instanceof Decoder) {
this.decoderImpl = (Decoder) o;
}
} catch (ClassNotFoundException cnfe) {
getLogger().warn("Unable to find decoder", cnfe);
} catch (IllegalAccessException iae) {
getLogger().warn("Could not construct decoder", iae);
} catch (InstantiationException ie) {
getLogger().warn("Could not construct decoder", ie);
}
try {
addr = InetAddress.getByName(address);
} catch (UnknownHostException uhe) {
uhe.printStackTrace();
}
try {
active = true;
socket = new MulticastSocket(port);
socket.joinGroup(addr);
receiverThread = new MulticastReceiverThread();
receiverThread.start();
handlerThread = new MulticastHandlerThread();
handlerThread.start();
if (advertiseViaMulticastDNS) {
zeroConf = new ZeroConfSupport(ZONE, port, getName());
zeroConf.advertise();
}
} catch (IOException ioe) {
ioe.printStackTrace();
}
}
public void setAdvertiseViaMulticastDNS(boolean advertiseViaMulticastDNS) {
this.advertiseViaMulticastDNS = advertiseViaMulticastDNS;
}
public boolean isAdvertiseViaMulticastDNS() {
return advertiseViaMulticastDNS;
}
class MulticastHandlerThread extends Thread {
private List list = new ArrayList();
public MulticastHandlerThread() {
setDaemon(true);
}
public void append(String data) {
synchronized (list) {
list.add(data);
list.notify();
}
}
public void run() {
ArrayList list2 = new ArrayList();
while (isAlive()) {
synchronized (list) {
try {
while (list.size() == 0) {
list.wait();
}
if (list.size() > 0) {
list2.addAll(list);
list.clear();
}
} catch (InterruptedException ie) {
}
}
if (list2.size() > 0) {
Iterator iter = list2.iterator();
while (iter.hasNext()) {
String data = (String) iter.next();
List v = decoderImpl.decodeEvents(data.trim());
if (v != null) {
Iterator eventIter = v.iterator();
while (eventIter.hasNext()) {
if (!isPaused()) {
doPost((LoggingEvent) eventIter.next());
}
}
}
}
list2.clear();
} else {
try {
synchronized (this) {
wait(1000);
}
} catch (InterruptedException ie) {
}
}
}
}
}
class MulticastReceiverThread extends Thread {
public MulticastReceiverThread() {
setDaemon(true);
}
public void run() {
active = true;
byte[] b = new byte[PACKET_LENGTH];
DatagramPacket p = new DatagramPacket(b, b.length);
while (active) {
try {
socket.receive(p);
//this string constructor which accepts a charset throws an exception if it is
//null
if (encoding == null) {
handlerThread.append(
new String(p.getData(), 0, p.getLength()));
} else {
handlerThread.append(
new String(p.getData(), 0, p.getLength(), encoding));
}
} catch (SocketException se) {
//disconnected
} catch (IOException ioe) {
ioe.printStackTrace();
}
}
getLogger().debug("{}'s thread is ending.", MulticastReceiver.this.getName());
}
}
}