slpdaemonimpl.java

Introduction to Sybase IQ.pdf

/* Copyright (c) 2005-2008 Jan S. Rellermeyer
 * Systems Group,
 * Department of Computer Science, ETH Zurich.
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *    - Redistributions of source code must retain the above copyright notice,
 *      this list of conditions and the following disclaimer.
 *    - Redistributions in binary form must reproduce the above copyright
 *      notice, this list of conditions and the following disclaimer in the
 *      documentation and/or other materials provided with the distribution.
 *    - Neither the name of ETH Zurich nor the names of its contributors may be
 *      used to endorse or promote products derived from this software without
 *      specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 */
package ch.ethz.iks.slp.impl;

import java.io.BufferedInputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.SortedMap;
import java.util.TreeMap;

import ch.ethz.iks.slp.ServiceLocationException;
import ch.ethz.iks.slp.ServiceType;
import ch.ethz.iks.slp.ServiceURL;

/**
 * the jSLP daemon class. This class is only required, if the peer is configured
 * as a SA and no other SLP daemon is running on the machine. UA-only
 * configurations or distributions that are intended to run on a machine with
 * OpenSLP <i>slpd</i> can be packaged without this class.
 * 
 * @author Jan S. Rellermeyer, ETH Zurich
 * @since 0.6
 */
public final class SLPDaemonImpl implements SLPDaemon {

	/**
	 * thread loop variable.
	 */
	private boolean running = true;

	/**
	 * Map of registered services:
	 * 
	 * String scope -> List of ServiceURLs services.
	 */
	private Map registeredServices = new HashMap();

	/**
	 * Sorted set for disposal of services which lifetimes have expired:
	 * 
	 * Long expirationTimestamp -> ServiceURL service.
	 */
	private SortedMap serviceDisposalQueue = new TreeMap();

	/**
	 * create a new SLPDaemon instance.
	 * 
	 * @param tcpSocket
	 *            the server socket.
	 * @throws Exception
	 *             if something goes wrong.
	 */
	public SLPDaemonImpl() throws Exception {
		new TcpServerThread();
		new ServiceDisposalThread();
		SLPCore.platform.logDebug("jSLP daemon starting ...");
	}

	/**
	 * register a service with the SLP framework. For the scopes, where DAs are
	 * known, the service will be registered with all DAs.
	 * 
	 * @param reg
	 *            the ServiceRegistration.
	 */
	private void registerService(final ServiceRegistration reg) {

		Service service = new Service(reg);

		for (Iterator scopeIter = reg.scopeList.iterator(); scopeIter.hasNext();) {
			String scope = (String) scopeIter.next();
			scope = scope.toLowerCase();
			synchronized (registeredServices) {
				SLPUtils.addValue(registeredServices, scope, service);
			}
			if (reg.url.getLifetime() > ServiceURL.LIFETIME_PERMANENT) {
				synchronized (serviceDisposalQueue) {
					long next = System.currentTimeMillis()
							+ (reg.url.getLifetime() * 1000);
					ArrayList keys = new ArrayList(serviceDisposalQueue
							.keySet());
					for (Iterator iter = keys.iterator(); iter.hasNext();) {
						Object key = iter.next();
						if (serviceDisposalQueue.get(key).equals(reg.url)) {
							serviceDisposalQueue.remove(key);
						}
					}
					serviceDisposalQueue.put(new Long(next), reg.url);
					serviceDisposalQueue.notifyAll();
				}
			}

			SLPCore.platform.logTraceReg("REGISTERED " + reg.url);

			// register the service with all known DAs in the scopes
			List daList = (List) SLPCore.dAs.get(scope);

			// no DA for the scope known ?
			// try to find one
			if ((daList == null || daList.isEmpty()) && !SLPCore.noDiscovery) {
				try {
					SLPCore.daLookup(Arrays
							.asList(new String[] { (String) scope }));

					// wait a short time for incoming replies
					synchronized (SLPCore.dAs) {
						try {
							SLPCore.dAs.wait(SLPCore.CONFIG.getWaitTime());
						} catch (InterruptedException e) {
						}
					}
					daList = (List) SLPCore.dAs.get(scope);
				} catch (ServiceLocationException sle) {
					SLPCore.platform.logError(sle.getMessage(), sle
								.fillInStackTrace());
				}
			}

			if (daList != null && !daList.isEmpty()) {
				final String[] dAs = (String[]) daList
						.toArray(new String[daList.size()]);
				final ServiceRegistration announcement = new ServiceRegistration(
						reg.url, reg.serviceType, reg.scopeList, reg.attList,
						reg.locale);
				announcement.authBlocks = reg.authBlocks;
				for (int i = 0; i < dAs.length; i++) {
					try {
						announceService(dAs[i], announcement);
						SLPCore.platform.logTraceReg("ANNOUNCED "
									+ announcement.url + " to " + dAs[i]);
					} catch (ServiceLocationException e) {
						// remove DA from list
						SLPUtils.removeValueFromAll(SLPCore.dAs, dAs[i]);
						SLPCore.dASPIs.remove(dAs[i]);
						SLPCore.platform.logError(e.getMessage(), e
									.fillInStackTrace());
					}
				}
			}
		}

	}

	/**
	 * deregister a service from the SLP framework. Deregisters from all DAs
	 * within the scopes and from the local service cache.
	 * 
	 * @param dereg
	 *            the service deregistration.
	 * @throws ServiceLocationException
	 */
	private void deregisterService(final ServiceDeregistration dereg)
			throws ServiceLocationException {

		final String[] scopes = (String[]) registeredServices.keySet().toArray(
				new String[registeredServices.size()]);
		for (int i = 0; i < scopes.length; i++) {
			final List tmp = (List) registeredServices.get(scopes[i]);
			final Service[] services = (Service[]) tmp.toArray(new Service[tmp
					.size()]);

			for (int j = 0; j < services.length; j++) {
				if (dereg.url.matches(services[j].url)) {
					List daList = (List) SLPCore.dAs.get(scopes[i].toLowerCase());
					if (daList != null) {
						for (Iterator daIter = daList.iterator(); daIter
								.hasNext();) {
							try {
								String dA = (String) daIter.next();
								dereg.address = InetAddress.getByName(dA);
								dereg.port = SLPCore.SLP_RESERVED_PORT;
								dereg.xid = SLPCore.nextXid();
								if (SLPCore.CONFIG.getSecurityEnabled()) {
									List spiList = (List) SLPCore.dASPIs
											.get(dA);
									dereg.sign(spiList);
								}
								ReplyMessage reply = SLPCore.sendMessage(dereg,
										true);
								if (reply.errorCode != 0) {
									throw new ServiceLocationException(
											(short) reply.errorCode,
											"Error during deregistration: "
													+ reply.errorCode);
								}
							} catch (UnknownHostException uhe) {
								throw new ServiceLocationException(
										ServiceLocationException.NETWORK_ERROR,
										uhe.getMessage());
							}
						}
					}
					synchronized (registeredServices) {
						SLPUtils.removeValue(registeredServices, scopes[i],
								services[j]);
					}
					break;
				}
			}
		}
	}

	/**
	 * all incoming messages are handled here.
	 * 
	 * @param msg
	 *            the message to be processed.
	 * @return the reply if the handled message came in via TCP. Otherwise null
	 *         will be returned.
	 * @throws ServiceLocationException
	 *             for various reasons like authentication failures etc.
	 */
	public ReplyMessage handleMessage(final SLPMessage msg)
			throws ServiceLocationException {
		if (msg == null) {
			return null;
		}

		String via = msg.tcp ? " (tcp)" : " (udp)";

		SLPCore.platform.logTraceMessage("RECEIVED (" + msg.address + ":"
					+ msg.port + ") " + msg.toString() + via);

		ReplyMessage reply = null;

		switch (msg.funcID) {
		case SLPMessage.SRVRQST:
			ServiceRequest req = (ServiceRequest) msg;

			List results = new ArrayList();
			for (Iterator scopes = req.scopeList.iterator(); scopes.hasNext();) {
				List services = (List) registeredServices.get(scopes.next());
				if (services == null) {
					continue;
				}

				for (Iterator srvs = services.iterator(); srvs.hasNext();) {
					Service service = (Service) srvs.next();
					if (service.url.getServiceType().matches(req.serviceType)) {
						if (req.predicate == null) {
							results.add(service.url);
							continue;
						}
						if (req.predicate.match(service.attributes)) {
							results.add(service.url);
						}
					}
				}
			}

			/*
			 * if there is no result, don't send a reply. This causes the SA to
			 * get the same message at least two more times but the RFC strictly
			 * demands this for multicast requests
			 */
			if (results.size() == 0 && req.multicast) {
				return null;
			}

			reply = new ServiceReply(req, results);

			if (SLPCore.CONFIG.getSecurityEnabled()) {
				((ServiceReply) reply).sign(req.spi);
			}

			return reply;