public MessageAdaptor sendAndWait(MessageAdaptor request) {
		String id = String.valueOf(counter());
		request.setId(id);
		messageSource.add(request);
		synchronized (publisher) {
			// logger.debug("State: " + publisher.getState());
			if (publisher.getState() == State.WAITING) {
				logger.info(String.format("Notify publisher %d.",
						publisher.getId()));
				publisher.notify();
			}
		}

		long current = System.currentTimeMillis();
		long limit = current + timeout;
		while (current < limit) {
			MessageAdaptor response = messageResponse.remove(id);
			if (response != null) {
				return response;
			}
			current = System.currentTimeMillis();
		}
//		logger.debug(""+messageResponse+" "+messageResponse.size());
		logger.error("sendAndWait timeout on id " + id);
		return null;
	}