slpdaemonimpl.java

Introduction to Sybase IQ.pdf

		case SLPMessage.ATTRRQST:
			AttributeRequest attreq = (AttributeRequest) msg;

			List attResult = new ArrayList();
			for (Iterator scopes = attreq.scopeList.iterator(); scopes
					.hasNext();) {
				List services = (List) registeredServices.get(scopes.next());
				if (services == null) {
					continue;
				}
				// the request can either be for a ServiceURL or a ServiceType
				Object reqService;
				boolean fullurl = false;
				if (attreq.url.indexOf("//") == -1) {
					reqService = new ServiceType(attreq.url);
				} else {
					fullurl = true;
					reqService = new ServiceURL(attreq.url, 0);
				}

				// if spi is sent, the request must be for a full url and
				// the tag list has to be empty
				if (attreq.spi.equals("")
						|| (fullurl && attreq.tagList.isEmpty())) {
					for (Iterator srvs = services.iterator(); srvs.hasNext();) {
						Service service = (Service) srvs.next();
						if (service.url.matches(reqService)) {
							attResult.addAll(SLPUtils.findMatches(
									attreq.tagList, service.attributes));
						}
					}
				}

			}
			reply = new AttributeReply(attreq, attResult);

			if (SLPCore.CONFIG.getSecurityEnabled()) {
				((AttributeReply) reply).sign(attreq.spi);
			}

			return reply;
		case SLPMessage.SRVTYPERQST:
			ServiceTypeRequest streq = (ServiceTypeRequest) msg;

			ArrayList result = new ArrayList();

			// iterate over scopes
			for (Iterator scopeIter = streq.scopeList.iterator(); scopeIter
					.hasNext();) {

				// iterate over the registered services
				List services = ((List) registeredServices
						.get(scopeIter.next()));
				if (services == null) {
					continue;
				}
				for (Iterator iter = services.iterator(); iter.hasNext();) {
					Service service = (Service) iter.next();
					ServiceType type = service.url.getServiceType();
					if (streq.namingAuthority.equals("*")
							|| streq.namingAuthority.equals("")
							|| type.getNamingAuthority().equals(
									streq.namingAuthority)) {
						if (!result.contains(type)) {
							result.add(type);
						}
					}
				}
			}
			reply = new ServiceTypeReply(streq, result);

			return reply;

		case SLPMessage.SRVREG:
			registerService((ServiceRegistration) msg);
			reply = new ServiceAcknowledgement(msg, 0);
			return reply;

		case SLPMessage.SRVDEREG:
			deregisterService((ServiceDeregistration) msg);

			reply = new ServiceAcknowledgement(msg, 0);

			return reply;

		case SLPMessage.SRVACK:
			final ReplyMessage rep = (ReplyMessage) msg;
			if (rep.errorCode != 0) {
				SLPCore.platform.logWarning(msg.address
							+ " replied with error code " + rep.errorCode
							+ " (" + rep + ")");
			}
			return null;

		default:
			// this should never happen, message should already cause an
			// exception during parsing
			throw new ServiceLocationException(
					ServiceLocationException.NOT_IMPLEMENTED,
					"The message type " + SLPMessage.getType(msg.funcID)
							+ " is not implemented");
		}

	}

	/**
	 * get informed about a new discovered DA. Registers all services in the
	 * scopes of the new DA.
	 * 
	 * @param advert
	 *            the DA advertisement.
	 */
	public void newDaDiscovered(final DAAdvertisement advert) {
		// so find all services within the scopes of the new DA:
		for (Iterator iter = advert.scopeList.iterator(); iter.hasNext();) {
			String scope = (String) iter.next();
			List services = (List) registeredServices.get(scope);
			if (services != null) {
				for (Iterator serviceIter = services.iterator(); serviceIter
						.hasNext();) {
					// and try to register it with the new DA
					try {
						Service service = (Service) serviceIter.next();
						ServiceRegistration reg = new ServiceRegistration(
								service.url, service.url.getServiceType(),
								Arrays.asList(new Object[] { scope }), SLPUtils
										.dictToAttrList(service.attributes),
								SLPCore.DEFAULT_LOCALE);
						SLPCore.platform.logDebug("Registering "
									+ service.url + " with new DA "
									+ advert.url);
						announceService(advert.url, reg);
					} catch (ServiceLocationException e) {
						SLPCore.platform.logError(e.getMessage(), e
									.fillInStackTrace());
					}
				}
			}
		}
	}

	/**
	 * register a service with a DA.
	 * 
	 * @param dAAddress
	 *            the IP address of the DA as <code>String</code>
	 * @param reg
	 *            the <code>ServiceRegistration</code> message.
	 * @throws ServiceLocationException
	 *             in case of network errors.
	 */
	private void announceService(final String dAAddress,
			final ServiceRegistration reg) throws ServiceLocationException {
		try {
			reg.address = InetAddress.getByName(dAAddress);
			reg.port = SLPCore.SLP_RESERVED_PORT;
			reg.xid = SLPCore.nextXid();
			if (SLPCore.CONFIG.getSecurityEnabled()) {
				List spiList = (List) SLPCore.dASPIs.get(dAAddress);
				reg.sign(spiList);
			}
			handleMessage(SLPCore.sendMessage(reg, true));
		} catch (UnknownHostException e) {
			SLPCore.platform.logError("Service announcement to "
						+ dAAddress + " failed. ", e.fillInStackTrace());
		}
	}

	/**
	 * TCP server thread.
	 */
	private final class TcpServerThread extends Thread {
		private ServerSocket socket;

		/**
		 * creates and starts a new TCP server thread.
		 * 
		 * @throws IOException
		 *             if socket creation fails.
		 */
		private TcpServerThread() throws IOException {
			socket = new ServerSocket(SLPCore.SLP_PORT);
			start();
		}

		/**
		 * thread loop.
		 */
		public void run() {
			while (running) {
				try {
					Socket con = socket.accept();
					DataInputStream in = new DataInputStream(
							new BufferedInputStream(con.getInputStream()));
					SLPMessage msg = SLPMessage.parse(con.getInetAddress(), con
							.getPort(), in, true);

					ReplyMessage reply = handleMessage(msg);
					if (reply != null) {
						SLPCore.platform.logTraceMessage("SEND REPLY ("
									+ reply.address + ":" + reply.port + ") "
									+ reply);

						DataOutputStream out = new DataOutputStream(con
								.getOutputStream());
						reply.writeTo(out);
						/*
						 * TODO the RFC encourages to keep the connection open
						 * to allow the other side to send multiple requests per
						 * connection. So start a server thread for every
						 * incoming connection instead of closing the connection
						 * after the first request
						 */
						out.close();
					}
					in.close();
					con.close();
				} catch (Exception ioe) {
					SLPCore.platform.logError(
								"Exception in TCP receiver thread", ioe);
				}
			}
		}
	}

	/**
	 * service disposal thread. Removes services from the local registry when
	 * their lifetime has expired.
	 */
	private final class ServiceDisposalThread extends Thread {

		/**
		 * create and start a new instance of this thread.
		 * 
		 */
		private ServiceDisposalThread() {
			start();
		}

		/**
		 * thread's main loop.
		 */
		public void run() {
			try {
				while (running) {
					synchronized (serviceDisposalQueue) {
						if (serviceDisposalQueue.isEmpty()) {
							// nothing to do, sleep until something arrives
							SLPCore.platform
										.logDebug("ServiceDisposalThread sleeping ...");
							serviceDisposalQueue.wait();
						} else {
							// we have work, do everything that is due
							Long nextActivity;
							while (!serviceDisposalQueue.isEmpty()
									&& (nextActivity = ((Long) serviceDisposalQueue
											.firstKey())).longValue() <= System
											.currentTimeMillis()) {
								ServiceURL service = (ServiceURL) serviceDisposalQueue
										.get(nextActivity);

								ServiceDeregistration dereg = new ServiceDeregistration(
										service, null, null,
										SLPCore.DEFAULT_LOCALE);
								try {
									deregisterService(dereg);
								} catch (ServiceLocationException sle) {
									SLPCore.platform.logError(sle
												.getMessage(), sle
												.fillInStackTrace());
								}
								SLPCore.platform
											.logTraceReg("disposed service "
													+ service);
								serviceDisposalQueue.remove(nextActivity);
							}
							if (!serviceDisposalQueue.isEmpty()) {
								/*
								 * there are some activities in the future,
								 * sleep until the first activity becomes due
								 */
								nextActivity = ((Long) serviceDisposalQueue
										.firstKey());
								long waitTime = nextActivity.longValue()
										- System.currentTimeMillis();
								if (waitTime > 0) {
									SLPCore.platform
												.logDebug("sleeping for "
														+ waitTime / 1000
														+ " seconds.");
									serviceDisposalQueue.wait(waitTime);
								}
							}
						}
					}
				}
			} catch (InterruptedException ie) {
				// let the thread stop.
			}
		}
	}
}