public class AISProcessor extends SignalkProcessor implements Processor {


	private static Logger logger = LogManager.getLogger(AISProcessor.class);

    /** Reader to parse lines and deliver complete AIS packets.
     * Updates them into model, and removes the key from the map. */
   
	
    private AISHandler aisHandler = new AISHandler(); 
	

	public void process(Exchange exchange) {
		if (exchange.getIn().getBody() == null)
			return;
		if (exchange.getIn().getBody() instanceof String){
				
			String bodyStr = exchange.getIn().getBody(String.class);
			try{
				String src = exchange.getIn().getHeader(MSG_SRC_BUS, String.class);
				SignalKModel model = aisHandler.handle(bodyStr, src);
				if(model!=null){
					exchange.getIn().setBody(model);
				}
			} catch (Exception e) {
				logger.debug(e.getMessage(),e);
				logger.error(e.getMessage()+" : "+bodyStr);
				exchange.getIn().setBody(bodyStr);
			}
			
			
		}

	}

   
} public class AisExpiryProcessor extends SignalkProcessor  implements Processor {

	private static Logger logger = LogManager.getLogger(AisExpiryProcessor.class);
	private ConcurrentMap<String, Long> map = new ConcurrentHashMap<>();
	
	
	public AisExpiryProcessor(){
		signalkModel.getEventBus().register(this);
		for(String key: signalkModel.getKeys()){
			processPath(key);
		}
	}
	
	@Override
	public void process(Exchange exchange) throws Exception {
		//event based, just triggers the checking
		//30 min ago.
		long millis = System.currentTimeMillis()-(30*60*1000);
		 Iterator<java.util.Map.Entry<String, Long>> itr = map.entrySet().iterator();
		while(itr.hasNext()){
			java.util.Map.Entry<String, Long> e = itr.next();
			if(millis > e.getValue()){
				//drop the entry
				signalkModel.put(e.getKey(), null, "self.ais.expiry");
				itr.remove();
			}
		}
	}
	
	  /**
     * @param pathEvent the path that was changed
     */
    @Subscribe
    public void recordEvent(PathEvent pathEvent) {
        if (pathEvent == null)
            return;

        String path = pathEvent.getPath();
        processPath(path);
        
    }

	private void processPath(String path) {
		if(StringUtils.isBlank(path)) return;
		if (path.startsWith(vessels + dot)) {
			int p1 = path.indexOf(vessels) + vessels.length() + 1;
			int pos = path.indexOf(".", p1);
			if (pos < 0)return;
			path = path.substring(0, pos);
			if(vessels_dot_self.equals(path))return;
			map.put(path, System.currentTimeMillis());
		}
	}

} public class AlarmProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(AlarmProcessor.class);
	
	private AlarmHandler alarmsHandler = new AlarmHandler();
	
	public void process(Exchange exchange) throws Exception {
		
		try {
			alarmsHandler.handle(signalkModel);
			
		} catch (Exception e) {
			logger.error(e);
		}
	}

} public class AnchorWatchProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(AnchorWatchProcessor.class);
	private static String zonesKey = vessels_dot_self_dot+nav_anchor_currentRadius+dot+meta+dot+zones;
	private static String radiusKey = vessels_dot_self_dot+nav_anchor_currentRadius;
	private static String latKey = vessels_dot_self_dot+nav_position_latitude;
	private static String lonKey = vessels_dot_self_dot+nav_position_longitude;
	private static String anchorLatKey = vessels_dot_self_dot+nav_anchor_position_latitude;
	private static String anchorLonKey = vessels_dot_self_dot+nav_anchor_position_longitude;
	public void process(Exchange exchange) throws Exception {
		
		try {
			Object zonesJson = signalkModel.get(zonesKey);
			if(zonesJson !=null && zonesJson instanceof Json && ((Json)zonesJson).isArray() && ((Json)zonesJson).asJsonList().size()>0){
				//anchor watch is on
				double lat = (double) signalkModel.get(latKey);
				double lon = (double) signalkModel.get(lonKey);
				double anchorLat = (double) signalkModel.get(anchorLatKey);
				double anchorLon = (double) signalkModel.get(anchorLonKey);
				//workout distance
				double distance = Util.haversineMeters(lat,lon,anchorLat, anchorLon);
				logger.debug("Updating anchor distance:"+distance);
				signalkModel.put(radiusKey,distance,null, Util.getIsoTimeString(System.currentTimeMillis()));
			}
			
		} catch (Exception e) {
			logger.error(e);
		}
	}

} public class ClientAppProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(ClientAppProcessor.class);

	private GitHandler handler = new GitHandler();

	@Override
	public void process(Exchange exchange) throws Exception {
		// the Restlet request should be available if neeeded
		HttpServletRequest request = exchange.getIn(HttpMessage.class).getRequest();
		HttpSession session = request.getSession();
		if (logger.isDebugEnabled())
			logger.debug("Session = " + session.getId());
		if (request.getSession() != null && allowInstall(request.getSession())) {
			if (request.getMethod().equals("GET")) {
				HttpServletResponse response = exchange.getIn(HttpMessage.class).getResponse();
				String path = request.getPathInfo();
				// String path = exchange.getIn().getHeader(Exchange.HTTP_URI, String.class);
				if (logger.isDebugEnabled())
					logger.debug("We are processing the path = " + path);
				// check addresses request
				if (path.startsWith(SignalKConstants.SIGNALK_INSTALL)) {
					handler.processInstall(request, response);
					return;
				}
				if (path.startsWith(SignalKConstants.SIGNALK_UPGRADE)) {
					handler.processUpgrade(request, response);
					return;
				}
				// response codes are set here, so all good now.
			}

		} else {
			HttpServletResponse response = exchange.getIn(HttpMessage.class).getResponse();
			response.setStatus(HttpServletResponse.SC_FORBIDDEN);
			response.sendRedirect(SignalKConstants.SIGNALK_AUTH);
			exchange.getIn().setBody(null);
		}

	}

	private boolean allowInstall(HttpSession session) {
		// TODO Need install security here
		return true;
	}

} public class ConfigFilterProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(ConfigFilterProcessor.class);
	JsonSerializer ser = new JsonSerializer();
	private boolean allow;
	
	public ConfigFilterProcessor(boolean allow){
		this.allow=allow;
	}
	
	public void process(Exchange exchange) throws Exception {
		
		if (exchange.getIn().getBody()==null)
			return;
		if(logger.isDebugEnabled())logger.debug("Processing, class="+exchange.getIn().getBody().getClass());
		//TODO: add more filters here
		if(exchange.getIn().getBody() instanceof Json){
			Json json = (Json)exchange.getIn().getBody();
			//remove _arduino
			try{
				if(allow){
					json.delAt(SignalKConstants.vessels);
				}else{
					json.delAt(SignalKConstants.CONFIG);
				}
			}catch(NullPointerException npe){}
			
			exchange.getIn().setBody(json.toString());
		}
		if (exchange.getIn().getBody() instanceof SignalKModel){
			SignalKModel model = (SignalKModel)exchange.getIn().getBody();
			//remove _arduino
			try{
				if(allow){
					model.getFullData().remove(SignalKConstants.vessels);
				}else{
					model.getFullData().remove(SignalKConstants.CONFIG);
				}
			}catch(NullPointerException npe){}
			
			exchange.getIn().setBody(ser.write(model));
		}
		if(logger.isDebugEnabled()){
			logger.debug("Outputting:"+exchange.getIn().getHeaders());
			logger.debug("Outputting:"+exchange.getIn());
		}
	}

} public class DeclinationProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(DeclinationProcessor.class);
	
	private DeclinationHandler decl = new DeclinationHandler();
	
	public void process(Exchange exchange) throws Exception {
		
		try {
		
			decl.handle(signalkModel);
			
		} catch (Exception e) {
			logger.error(e);
		}
	}

} public class DeltaImportProcessor extends SignalkProcessor implements Processor{

	private static Logger logger = LogManager.getLogger(DeltaImportProcessor.class);
	
	private DeltaToMapConverter deltaToMap = new DeltaToMapConverter();
	
	public void process(Exchange exchange) throws Exception {
		
		try {
			if(exchange.getIn().getBody()==null ||!(exchange.getIn().getBody() instanceof Json)) return;
			
			SignalKModel model = deltaToMap.handle(exchange.getIn().getBody(Json.class));
			
			if(model!=null){
				if(logger.isDebugEnabled())logger.debug("Converted to:"+model);
				exchange.getIn().setBody(model);
			}else{
				if(logger.isDebugEnabled())logger.debug("Ignored, not update:"+exchange.getIn().getBody(Json.class));
			}
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
		}
	}



	
} public class DepthProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(DepthProcessor.class);
	

	private DepthHandler depth = new DepthHandler();

	

	public void process(Exchange exchange) throws Exception {		
		try {
		
			depth.handle(signalkModel);
			
		} catch (Exception e) {
			logger.error(e);
		}
	}

	

} public class FullExportProcessor extends SignalkProcessor implements Processor, FullExportProcessorMBean {

	private static final Logger logger = LogManager.getLogger(FullExportProcessor.class);
    private static final Timer timer = new Timer("Export Timer", true);

    private final String wsSession;
    private final String routeId;
    private final AtomicLong lastSend;
    //private ThreadLocal<ConcurrentLinkedQueue<String>> pendingThreadLocal = new ThreadLocal<>();
    private final ConcurrentLinkedQueue<String> pendingPaths = new ConcurrentLinkedQueue<>();

    public FullExportProcessor(String wsSession, String routeId) {
        super();
        this.wsSession = wsSession;
        this.routeId=routeId;
        lastSend = new AtomicLong(System.currentTimeMillis());
        signalkModel.getEventBus().register(this);
        
       
        //setup jmx
        //register the MBean
		try {
			ObjectName name = new ObjectName(getClass().getName(),"id",UUID.randomUUID().toString());
			ManagementFactory.getPlatformMBeanServer().registerMBean(this, name);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			logger.error(e.getMessage(),e);
		} 
        
    }
    
	
	 public void process(Exchange exchange) throws Exception {
	        try {
	            if (logger.isDebugEnabled())
	                logger.info("process  subs for " + routeId + " as delta? " + isDelta(routeId));

	            // Clear the pending paths as we are about to send all.
	            
	            pendingPaths.clear();
	            lastSend.set(System.currentTimeMillis());

	            // get the accumulated delta nodes.
	            exchange.getIn().setBody(createTree(routeId));
	            setHeaders(exchange);
	            if (logger.isDebugEnabled()) {
	                logger.debug("Headers set to :" + exchange.getIn().getHeaders());
	               // logger.debug("Body set to :" + StringUtils.abbreviate(exchange.getIn().getBody().toString(),500));
	                logger.debug("Body set to :" + exchange.getIn().getBody().toString());
	            }
	        } catch (Exception e) {
	            logger.error(e.getMessage(), e);
	            throw e;
	        }
	    }
	/*public void process(Exchange exchange) throws Exception {

		try {
			if(logger.isInfoEnabled())logger.info("process  subs for " + exchange.getFromRouteId()+" as delta? "+isDelta(exchange.getFromRouteId()));
			// get the accumulated delta nodes.
			exchange.getIn().setBody(createTree(exchange.getFromRouteId()));
			setHeaders(exchange);
			if(logger.isDebugEnabled()){
				logger.debug("Headers set to :" + exchange.getIn().getHeaders());
				logger.debug("Body set to :" + exchange.getIn().getBody());
			}
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
			throw e;
		}
	}*/

	 private void setHeaders(Exchange exchange) {
	        for (Subscription sub : manager.getSubscriptions(wsSession)) {
	            if (sub == null || !sub.isActive() || !routeId.equals(sub.getRouteId()))
	                continue;
	            exchange.getIn().setHeader(SIGNALK_FORMAT, sub.getFormat());
	            if (sub.getDestination() != null) {
	                exchange.getIn().setHeader(ConfigConstants.DESTINATION, sub.getDestination());
	            }
	            exchange.getIn().setHeader(ConfigConstants.OUTPUT_TYPE, manager.getOutputType(sub.getWsSession()));
	            exchange.getIn().setHeader(WebsocketConstants.CONNECTION_KEY, sub.getWsSession());

	        }
	    }

	   
	 private boolean isDelta(String routeId) {
	        for (Subscription sub : manager.getSubscriptions(wsSession)) {
	            if (sub != null && sub.isActive() && routeId.equals(sub.getRouteId()) 
	            		&& FORMAT_DELTA.equals(sub.getFormat())) {
	                return true;
	            }
	        }
	        return false;
	    }

	   private SignalKModel createTree(String routeId) {
	        SignalKModel temp = SignalKModelFactory.getCleanInstance();
	        if(logger.isDebugEnabled())logger.debug("subs for ws:" + wsSession + " = " + manager.getSubscriptions(wsSession));
	        for (Subscription sub : manager.getSubscriptions(wsSession)) {
	            if (sub != null && sub.isActive() && routeId.equals(sub.getRouteId())) {
	            	if(logger.isDebugEnabled())logger.debug("Found active sub:" + sub);
	                for (String p : sub.getSubscribed(null)) {
	                    NavigableMap<String, Object> node = signalkModel.getSubMap(p);
	                    if(logger.isDebugEnabled())logger.debug("Found node:" + p + " = " + node);
	                    for (String key:node.keySet()){
	                    	if(key.contains(".meta."))continue;
	                    	if(key.contains(".values."))continue;
	                    	//if(key.contains(".source"))continue;
	                    	//if(key.contains(".$source"))continue;
	                    	Object val = node.get(key);
	                    	if(val!=null){
	                    		temp.getData().put(key,val);
	                    	}
	                    }
	                    
	                }
	            }
	        }
	        return temp;
	    }
/*
	private SignalKModel createTree(String routeId) {
		// we need the routeId.
		// Multimap<String,String> vesselList = TreeMultimap.create();
		SignalKModel temp = SignalKModelFactory.getCleanInstance();
		
		for (Subscription sub : manager.getSubscriptions(wsSession)) {
			if (sub == null || !sub.isActive() || !routeId.equals(sub.getRouteId()))
				continue;
			for (String p : sub.getSubscribed(null)) {
				populateTree(temp, p);
			}
		}
		return temp;

	}*/

	
	

	  /**
     * @param pathEvent the path that was changed
     */
    @Subscribe
    public void recordEvent(PathEvent pathEvent) {
        if (pathEvent == null)
            return;

        String path = pathEvent.getPath();
        if (path == null)
            return;
        
        if(path.endsWith(dot+source)
    			|| path.endsWith(dot+timestamp)
    			|| path.contains(dot+source+dot)
    			|| path.endsWith(dot+sourceRef)){
        	return;
        }
        // Send update if necessary.
        for (Subscription s : manager.getSubscriptions(wsSession)) {
            if (s.isActive() && s.isSubscribed(path) && routeId.equals(s.getRouteId())) {
                switch (s.getPolicy()) {
                    case POLICY_INSTANT:
                        // Always send now.
                        send(Collections.singletonList(path),s);
                        return;
                    case POLICY_IDEAL:
                        // Schedule a timer if the queue is currently empty.
                    	
                        boolean schedule = pendingPaths.isEmpty();
                        pendingPaths.add(path);

                        if (schedule) {
                            TimerTask task = new TimerTask() {
                                @Override
                                public void run() {
                                    List<String> paths = new ArrayList<>();
                                    for (String path = pendingPaths.poll(); path != null; path = pendingPaths.poll()) {
                                        paths.add(path);
                                    }
                                    send(paths,s);
                                }
                            };
                            // Schedule to send minPeriod after the last send; if that's in the past the task will run immediately.
                            timer.schedule(task, new Date(lastSend.get() + s.getMinPeriod()));
                        }
                        return;
                    case POLICY_FIXED:
                    default:
                        // Updates will be sent at the next period.
                        break;
                }
            }
        }
    }

    /**
	 * @param pathEvent
	 */
	/*@Subscribe
	public void recordEvent(PathEvent pathEvent) {
		if (pathEvent == null)
			return;
		if (pathEvent.getPath() == null)
			return;
		//if (logger.isTraceEnabled())logger.trace(this.wsSession + " received event " + pathEvent.getPath());
		if(logger.isDebugEnabled())logger.debug(this.wsSession + " received event " + pathEvent.getPath());
		// do we care?
		for (Subscription s : manager.getSubscriptions(wsSession)) {
			if (s.isActive() && !POLICY_FIXED.equals(s.getPolicy()) && s.isSubscribed(pathEvent.getPath())) {
				if(logger.isDebugEnabled())logger.debug("Adding to send queue : "+pathEvent.getPath());
				queue.add(pathEvent.getPath());
				sender.startSender();
				break;
			}
		}

	}*/
	@Subscribe
	public void recordEvent(DeadEvent e) {
		logger.debug("Received dead event" + e.getSource());
	}


    private void send(Collection<String> paths, Subscription sub) {
        if (paths.isEmpty()) {
            return;
        }

        SignalKModel temp = SignalKModelFactory.getCleanInstance();
        boolean send = false;
        for (String path : paths) {
            Object node = signalkModel.get(path);
            if (node != null) {
                temp.getFullData().put(path, node);
                send = true;
            }
        }

        if (send) {
            if (logger.isDebugEnabled()) {
                logger.debug("Sending : " + temp.getKeys());
            }

            Map<String, Object> headers = new HashMap<>();
            headers.put(WebsocketConstants.CONNECTION_KEY, wsSession);
            headers.put(ConfigConstants.OUTPUT_TYPE, manager.getOutputType(wsSession));
            if (sub.getDestination() != null) {
            	headers.put(ConfigConstants.DESTINATION, sub.getDestination());
            }
            asyncSendBodyAndHeaders(outProducer,temp, headers);
            lastSend.set(System.currentTimeMillis());
        } else {
            if (logger.isDebugEnabled()) {
                logger.debug("Nothing to send");
            }
        }

    }


	@Override
	public String getSubscriptionDetails() {
		ArrayList<String> subDetails = new ArrayList<>();
		for (Subscription s : manager.getSubscriptions(wsSession)) {
			if(routeId.equals(s.getRouteId())){
				subDetails.add(s.toString());
			}
		}
		return subDetails.toString();
	}
	
/*	class MsgSender implements Runnable {
		long lastSend = 0;

		Thread t = null;

		public void startSender() {
			if(logger.isDebugEnabled())logger.debug("Checking sender..");
			if (t != null && t.isAlive())
				return;
			if(logger.isDebugEnabled())logger.debug("Starting sender..");
			t = new Thread(this);
			t.start();
		}

		@Override
		public void run() {
			// TODO make this react to actual subs minPeriod
			while (true) {
				if (System.currentTimeMillis() - lastSend > 100) {
					// send messages
					String p = null;
					SignalKModel temp = SignalKModelFactory.getCleanInstance();
					boolean send = false;
					while ((p = queue.poll()) != null) {
						Object node = signalkModel.get(p);
						if(logger.isDebugEnabled())logger.debug("Found node:" + p + " = " + node);
						if (node != null) {
							temp.getFullData().put(p, node);
							send = true;
						}
					}
					if (send) {
						HashMap<String, Object> headers = new HashMap<String, Object>();
						headers.put(WebsocketConstants.CONNECTION_KEY, wsSession);
						headers.put(ConfigConstants.OUTPUT_TYPE, manager.getOutputType(wsSession));
						asyncSendBodyAndHeaders(outProducer,temp, headers);
						lastSend = System.currentTimeMillis();
					}
					break;
				} else {
					try {
						Thread.sleep(20);
					} catch (InterruptedException e) {
					}
				}
			}

		}

		

	}*/

} public class FullImportProcessor extends SignalkProcessor implements Processor{

	private static Logger logger = LogManager.getLogger(FullImportProcessor.class);
	
	private FullToMapConverter fullToMap = new FullToMapConverter();
	
	public void process(Exchange exchange) throws Exception {
		
		try {
			if(exchange.getIn().getBody()==null ||!(exchange.getIn().getBody() instanceof Json)) return;
			//we only process full format
			Json node = exchange.getIn().getBody(Json.class);
			if(node.has(vessels) || node.has(CONFIG) || node.has(resources)){
				SignalKModel model = fullToMap.handle(node);
				if(logger.isDebugEnabled())logger.debug("Converted to:"+model);
				if(model!=null){
					exchange.getIn().setBody(model);
				}
			}else{
				if(logger.isDebugEnabled())logger.debug("Ignored, not update:"+exchange.getIn().getBody(Json.class));
			}
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
		}
	}



	
} public class FullToDeltaProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(FullToDeltaProcessor.class);

	private FullToDeltaConverter fullToDelta = new FullToDeltaConverter();
	
	public void process(Exchange exchange) throws Exception {

		try {
			if (exchange.getIn().getBody() == null) return;
			if(logger.isDebugEnabled())logger.debug("Processing :" + exchange.getIn().getBody().toString());
			if(!(exchange.getIn().getBody() instanceof Json))return;
			String format = exchange.getIn().getHeader(SIGNALK_FORMAT, String.class);
			if (StringUtils.isBlank(format) || FORMAT_DELTA.equals(format)) {
				List<Json> json = fullToDelta.handle(exchange.getIn().getBody(Json.class));
				//if(logger.isDebugEnabled())logger.debug("Converted to delta :" + json);
				exchange.getIn().setBody(json);
			}
			if (exchange.getIn().getBody() == null) return;
			if(logger.isDebugEnabled())logger.debug("Outputting :" + exchange.getIn().getBody().toString());
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

} public class HeartbeatProcessor extends SignalkProcessor implements Processor{

	private static Logger logger = LogManager.getLogger(HeartbeatProcessor.class);
	private SubscriptionManager manager = SubscriptionManagerFactory.getInstance();
	ProducerTemplate producer;
	//Json msg = null;
	
	public HeartbeatProcessor(){
		producer= new DefaultProducerTemplate(CamelContextFactory.getInstance());
		producer.setDefaultEndpointUri(RouteManager.SEDA_COMMON_OUT );
		try {
			producer.start();
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
		}
		
	}
	
	public void process(Exchange exchange) throws Exception {
		
		try {
			List<String> heartbeats = ImmutableList.copyOf(manager.getHeartbeats());
			if(logger.isDebugEnabled())logger.debug("process heartbeats: "+heartbeats.size());
			Json msg=getHeartBeatDelta();
			for(String session : heartbeats){
				HashMap<String, Object> headers = new HashMap<String, Object>();
				headers.put(WebsocketConstants.CONNECTION_KEY, session);
				headers.put(ConfigConstants.OUTPUT_TYPE, manager.getOutputType(session));
				asyncSendBodyAndHeaders(producer,msg, headers);
			}
			
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
			throw e;
		}
	}
	
	
	/**
	 *Should return the basic hello message
	 *<pre> {
	*	  "version": "0.1",
	*	  "timestamp": "2015-04-13T01:13:50.524Z",
	*	  "self": "123456789"
	*	}
	*</pre>
	 * @return
	 */
	private Json getHeartBeatDelta(){
		Json delta = Json.object();
		delta.set(SignalKConstants.version, Util.getConfigProperty(SignalKConstants.version));
		delta.set(SignalKConstants.timestamp, Util.getIsoTimeString());
		delta.set(SignalKConstants.self_str, Util.getConfigProperty(SignalKConstants.self));		
		return delta;
	}
	
} public class IncomingSecurityFirewall extends SignalkProcessor implements
		Processor {

	private static Logger logger = LogManager
			.getLogger(IncomingSecurityFirewall.class);
	private List<String> whiteList = new ArrayList<String>();
	private List<String> configAcceptList = new ArrayList<String>();
	private List<String> denyList = new ArrayList<String>();

	public IncomingSecurityFirewall() {
		// load lists now
		Json deny = Util.getConfigJsonArray(ConfigConstants.SECURITY_DENY);
		Json white = Util.getConfigJsonArray(ConfigConstants.SECURITY_WHITE);
		Json config = Util.getConfigJsonArray(ConfigConstants.SECURITY_CONFIG);
		if (deny != null) {
			for (Object o : deny.asList()) {
				denyList.add((String) o);
			}
		}
		if (white != null) {
			for (Object o : white.asList()) {
				whiteList.add((String) o);
			}
		}
		if (config != null) {
			for (Object o : config.asList()) {
				configAcceptList.add((String) o);
			}
		}
	}

	public void process(Exchange exchange) throws Exception {

		try {
			// we trust local serial
			String type = exchange.getIn().getHeader(MSG_TYPE,
					String.class);
			if (SERIAL.equals(type)||DEMO.equals(type)){
				if (logger.isDebugEnabled())
					logger.debug("Accepting msg:" + type);
				return;
			}
			//TODO: check for valid XMPP/STOMP/MQTT SRC_IP ??
			
			// we filter on ip
			String srcIp = exchange.getIn().getHeader(MSG_SRC_IP,
					String.class);
			if (logger.isDebugEnabled())
				logger.debug("Checking src ip:" + srcIp);
			if (srcIp == null) {
				logger.debug("Src ip null:"+exchange.getIn().getHeaders());
				//exchange.getIn().setBody(null);
				return;
			}
			// denied - drop now
			//loop and check
			if(Util.inNetworkList(denyList, srcIp)){
				if (logger.isDebugEnabled())
					logger.warn("Message DENIED for src ip :" + srcIp);
				exchange.getIn().setBody(null);
				return;
			}

			// save config only from allowed ips
			String configSave = exchange.getIn().getHeader(
					CONFIG_ACTION, String.class);
			if (CONFIG_ACTION_SAVE.equals(configSave)) {
				// must be in the configAcceptlist!
				
				if(Util.inNetworkList(configAcceptList, srcIp)){
					if (logger.isDebugEnabled())
						logger.debug("Config save allowed for src ip:" + srcIp);
					return;
				} else {
					if (logger.isDebugEnabled())
						logger.warn("Config save DENIED for src ip:" + srcIp);
					exchange.getIn().setBody(null);
					return;
				}
			}
			// we trust INTERNAL_IP
			if (INTERNAL_IP.equals(type)) {
				if (logger.isDebugEnabled())
					logger.debug("Message allowed for src ip (internal):"+ srcIp+":"+exchange.getIn().getHeaders());
				return;
			}

			// we trust our whitelist
			if(Util.inNetworkList(whiteList, srcIp)){
				if (logger.isDebugEnabled())
					logger.debug("WhiteList message allowed for src ip:" + srcIp);
				return;
			}

			// now we look for anomalies

			// new incoming, so flag for acceptance
			// exchange.getIn().setHeader(MSG_APPROVAL,
			// REQUIRED);
			// filter for evil
			Json node = exchange.getIn().getBody(Json.class);

			if (node.at(UPDATES) != null
					|| node.at(PUT) != null) {
				// cant be an update or put for this vessel since its external
				if (node.at(CONTEXT).asString()
						.contains(self)) {
					if (logger.isDebugEnabled())
						logger.debug("Message DENIED for src ip (spoofing self):"
								+ srcIp);
					exchange.getIn().setBody(null);
					return;
				}
			}
			// filter(node);
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

	public void filter(Json node) {
		// apply rules to this object

		// recurse into object
		for (Json n : node.asJsonMap().values()) {
			filter(n);
		}

	}

} public class InputFilterProcessor extends SignalkProcessor implements Processor {
	private static Logger logger = LogManager.getLogger(InputFilterProcessor.class);

	public InputFilterProcessor() {

	}

	public void process(Exchange exchange) throws Exception {
		String msg = exchange.getIn().getBody(String.class);
		if (msg != null) {
			msg = msg.trim();
			// stomp messages are prefixed with 'ascii:'
			if (msg.startsWith("ascii:"))
				msg = msg.substring(6).trim();
			msg = StringUtils.chomp(msg);
			boolean ok = false;
			if (msg.startsWith("!AIVDM")) {
				// AIS
				// !AIVDM,1,1,,B,15MwkRUOidG?GElEa<iQk1JV06Jd,0*6D
				exchange.getIn().setBody(msg);
				sendNmea(exchange);
				ok = true;
			} else if (msg.startsWith("$")) {
				// NMEA - good
				// System.out.println(msg);
				exchange.getIn().setBody(msg);
				if(logger.isDebugEnabled())
						logger.debug(msg.toString());
                sendNmea(exchange);
				ok = true;
			} else if (msg.startsWith("{") && msg.endsWith("}")) {
				Json json = Json.read(msg);
				// n2k
				if (json.has(SignalKConstants.pgn)) {
					exchange.getIn().setHeader(SignalKConstants.N2K_MESSAGE, msg);
				}
				// full or delta format
				if (exchange.getIn().getHeader(SignalKConstants.SIGNALK_FORMAT) == null) {
					if (json.has(SignalKConstants.CONTEXT)) {
						exchange.getIn().setHeader(SignalKConstants.SIGNALK_FORMAT, SignalKConstants.FORMAT_DELTA);
					}
					if (json.has(SignalKConstants.vessels)||json.has(SignalKConstants.resources)) {
						exchange.getIn().setHeader(SignalKConstants.SIGNALK_FORMAT, SignalKConstants.FORMAT_FULL);
					}
				}
				// compensate for MQTT and STOMP sessionid
				if (json.has(WebsocketConstants.CONNECTION_KEY)) {
					exchange.getIn().setHeader(WebsocketConstants.CONNECTION_KEY, json.at(WebsocketConstants.CONNECTION_KEY).asString());
				}
				// deal with REPLY_TO
				Map<String, Object> headers = exchange.getIn().getHeaders();
				if (headers != null && headers.containsKey(ConfigConstants.REPLY_TO)) {
					exchange.getIn().setHeader(ConfigConstants.DESTINATION, headers.get(ConfigConstants.REPLY_TO));
					// headers.remove(Constants.REPLY_TO);
				}
				// for MQTT
				if (json.has(ConfigConstants.REPLY_TO)) {
					exchange.getIn().setHeader(ConfigConstants.DESTINATION, (json.at(ConfigConstants.REPLY_TO).asString()));
				}
				//if it has a config object, flag it as such
				if (json.has(SignalKConstants.CONFIG)
						||(json.has(SignalKConstants.CONTEXT) && StringUtils.startsWith(json.at(SignalKConstants.CONTEXT).toString(),SignalKConstants.CONFIG))) {
					exchange.getIn().setHeader(SignalKConstants.CONFIG_ACTION, SignalKConstants.CONFIG_ACTION_SAVE);
				}
				// json
				exchange.getIn().setBody(json);
				ok = true;
			}
			if (ok) {
				return;
			}
			// uh-oh log it, squash it
			exchange.getUnitOfWork().done(exchange);
			// System.out.println("Dropped invalid message:"+msg);
			logger.info("Dropped invalid message:" + msg);
			exchange.getIn().setBody(null);
			// throw new CamelExchangeException("Invalid msg", exchange);
		}

	}

} public class JsonGetProcessor extends SignalkProcessor implements Processor{

	private static Logger logger = LogManager.getLogger(JsonGetProcessor.class);
	//private static DateTimeFormatter fmt = ISODateTimeFormat.dateTime();
	private JsonGetHandler handler = new JsonGetHandler();
	JsonSerializer ser = new JsonSerializer();
	public void process(Exchange exchange) throws Exception {
		
		try {
			if(exchange.getIn().getBody()==null ||!(exchange.getIn().getBody() instanceof Json)) return;
			String wsSession = exchange.getIn().getHeader(WebsocketConstants.CONNECTION_KEY, String.class);
			if(wsSession==null){
				if(logger.isDebugEnabled())logger.debug("Skipped, no session:"+exchange.getIn().getBody(Json.class));
				return;
			}
			Json json = exchange.getIn().getBody(Json.class);
			//avoid full signalk syntax
			if(json.has(vessels))return;
			if(json.has(CONTEXT) && (json.has(GET))){
				SignalKModel temp = handler.handle(signalkModel, json);
				if(logger.isDebugEnabled())logger.debug("Retrieved:"+temp);
				//also STOMP headers etc, swap replyTo
				Map<String, Object> headers = exchange.getIn().getHeaders();
				headers.put(SIGNALK_FORMAT, handler.getFormat(json));
				
				json.delAt(GET);
				//if(logger.isDebugEnabled())logger.debug("headers:"+headers);
				
				
				if(exchange.getIn().getHeader(RestApiProcessor.REST_REQUEST)!=null){
					//deal with GET
					//if(exchange.getIn().getHeader(RestApiProcessor.REST_WILDCARD)!=null
					//		&& exchange.getIn().getHeader(RestApiProcessor.REST_WILDCARD).equals("false")){
						//then we only send back the object
						
					//}
					exchange.getIn().setBody(ser.writeJson(temp));
					if(logger.isDebugEnabled())logger.debug("Processed REST GET request:"+exchange.getIn().getBody(Json.class));
				}else{
					if(logger.isDebugEnabled())logger.debug("Processed GET request:"+temp);
					asyncSendBodyAndHeaders(outProducer,temp, headers);
					//outProducer.sendBodyAndHeaders(temp, headers);
				}
			}else{
				if(logger.isDebugEnabled())logger.debug("Skipped, not a GET request:"+exchange.getIn().getBody(Json.class));
			}
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
		}
	}


	

	
} public class JsonListProcessor extends SignalkProcessor implements Processor{

	private static Logger logger = LogManager.getLogger(JsonListProcessor.class);
	//private static DateTimeFormatter fmt = ISODateTimeFormat.dateTime();
	private JsonListHandler listHandler = new JsonListHandler();
	
	
	
	public void process(Exchange exchange) throws Exception {
		
		try {
			if(exchange.getIn().getBody()==null ||!(exchange.getIn().getBody() instanceof Json)) return;
			String wsSession = exchange.getIn().getHeader(WebsocketConstants.CONNECTION_KEY, String.class);
			if(wsSession==null){
				if(logger.isDebugEnabled())logger.debug("Skipped, no session:"+exchange.getIn().getBody(Json.class));
				return;
			}
			Json json = exchange.getIn().getBody(Json.class);
			//avoid full signalk syntax
			if(json.has(vessels))return;
			if(json.has(CONTEXT) && (json.has(LIST))){
				Json pathList = listHandler.handle(json);
				json.set(PATHLIST, pathList);
				
				//also STOMP headers etc, swap replyTo
				Map<String, Object> headers = exchange.getIn().getHeaders();
				
				json.delAt(LIST);
				if(exchange.getIn().getHeader(RestApiProcessor.REST_REQUEST)!=null){
					if(logger.isDebugEnabled())logger.debug("Processed REST LIST request:"+exchange.getIn().getBody(Json.class));
					exchange.getIn().setBody(json);
				}else{
					if(logger.isDebugEnabled())logger.debug("Processed LIST request:"+exchange.getIn().getBody(Json.class));
					asyncSendBodyAndHeaders(outProducer,json, headers);
					//outProducer.sendBodyAndHeaders(json, headers);
				}
				
			}else{
				if(logger.isDebugEnabled())logger.debug("Skipped, not a LIST request:"+exchange.getIn().getBody(Json.class));
			}
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
		}
	}

		
} public class JsonSubscribeProcessor extends SignalkProcessor implements Processor{

	private static Logger logger = LogManager.getLogger(JsonSubscribeProcessor.class);
	//private static DateTimeFormatter fmt = ISODateTimeFormat.dateTime();
	
	public void process(Exchange exchange) throws Exception {
		
		try {
			if(exchange.getIn().getBody()==null ||!(exchange.getIn().getBody() instanceof Json)) return;
			String wsSession = exchange.getIn().getHeader(WebsocketConstants.CONNECTION_KEY, String.class);
			if(wsSession==null){
				if(logger.isDebugEnabled())logger.debug("WsSession is null:"+exchange.getIn().getHeaders());
				//return;
			}
			Json json = exchange.getIn().getBody(Json.class);
			//avoid full signalk syntax
			if(json.has(vessels))return;
			if(json.has(CONTEXT) && (json.has(SUBSCRIBE) || json.has(UNSUBSCRIBE))){
				json = handle(json, exchange.getIn().getHeaders());
				exchange.getIn().setBody(json);
			}else{
				if(logger.isDebugEnabled()){
					logger.debug("Subscribe msg class: "+exchange.getIn().getBody().getClass());
					logger.debug("Subscribe msg: "+exchange.getIn().getBody());
				}
			}
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
		}
	}

	/*
	 *  
	 *  <pre>
{
    "context": "vessels.*",
    "subscribe": [
        {
                    "path": "navigation.position",
                    "period": 120000,
                    "format": "full",
                    "policy": "fixed"
                },
         {
                   "path": "navigation.courseOverGround",
                    "period": 120000,
                    "format": "full",
                    "policy": "fixed"
                }
        ]
}

{"context":"vessels.*","subscribe":[{"path":"navigation.position","period":120000,"format":"full","policy":"fixed"},{"path":"navigation.courseOverGround","period":120000,"format":"full","policy":"fixed"}]}
</pre>
*/
	 
	//@Override
	public Json  handle(Json node, Map<String, Object> headers) throws Exception {
		
		//deal with diff format
		
		if(logger.isDebugEnabled())logger.debug("Checking for subscribe  "+node );
		
		//go to context
		String context = node.at(CONTEXT).asString();
		//TODO: is the context and path valid? A DOS attack is possible if we allow numerous crap/bad paths?
		
		Json subscriptions = node.at(SUBSCRIBE);
		if(subscriptions!=null){
			//MQTT and STOMP wont have created proper session links
			if(node.has(WebsocketConstants.CONNECTION_KEY)){
				String wsSession = node.at(WebsocketConstants.CONNECTION_KEY).asString();
				if(node.has(ConfigConstants.OUTPUT_TYPE)){
					String outputType = node.at(ConfigConstants.OUTPUT_TYPE).asString();
					SubscriptionManagerFactory.getInstance().add(wsSession, wsSession, outputType,"127.0.0.1","127.0.0.1");
				}
			}
			if(subscriptions.isArray()){
				for(Json subscription: subscriptions.asJsonList()){
					
					parseSubscribe(headers, context, subscription);
				}
			}
			if(logger.isDebugEnabled())logger.debug("processed subscribe  "+node );
		}
		Json unsubscriptions = node.at(UNSUBSCRIBE);
		if(unsubscriptions!=null){
		
			if(unsubscriptions.isArray()){
				for(Json subscription: unsubscriptions.asJsonList()){
					parseUnsubscribe(headers, context, subscription);
				}
			}
			if(logger.isDebugEnabled())logger.debug("processed unsubscribe  "+node );
		}
			
		
		return node;
		
	}

	/**
	 *  
	 *   <pre>{
                    "path": "navigation.speedThroughWater",
                    "period": 1000,
                    "format": "delta",
                    "policy": "ideal",
                    "minPeriod": 200
                }
                </pre>
	 * @param context
	 * @param subscription
	 * @throws Exception 
	 */
	private void parseSubscribe(Map<String, Object> headers, String context, Json subscription) throws Exception {
		//get values
		if(logger.isDebugEnabled())logger.debug("Parsing subscribe  "+subscription );
		String wsSession = (String) headers.get(WebsocketConstants.CONNECTION_KEY);
		String path = context+"."+subscription.at(PATH).asString();
		long period = 1000;
		if(subscription.at(PERIOD)!=null)period = subscription.at(PERIOD).asInteger();
		String format = FORMAT_DELTA;
		if(subscription.at(FORMAT)!=null)format=subscription.at(FORMAT).asString();
		String policy = POLICY_FIXED;
		if(subscription.at(POLICY)!=null)policy=subscription.at(POLICY).asString();
		long minPeriod = 0;
		if(subscription.at(MIN_PERIOD)!=null)minPeriod=subscription.at(MIN_PERIOD).asInteger();
		Subscription sub = new Subscription(wsSession, path, period, minPeriod, format, policy);
		
		//STOMP, MQTT
		if(headers.containsKey(ConfigConstants.DESTINATION)){
			sub.setDestination( headers.get(ConfigConstants.DESTINATION).toString());
		}
		
		//sub.setActive(false);
		if(logger.isDebugEnabled())logger.debug("Created subscription; "+sub.toString() );
		SubscriptionManagerFactory.getInstance().addSubscription(sub);
		
	}
	
	/**
	 *  
	 *   <pre>{
                    "path": "navigation.speedThroughWater",
                    "period": 1000,
                    "format": "delta",
                    "policy": "ideal",
                    "minPeriod": 200
                }
                </pre>
	 * @param context
	 * @param subscription
	 * @throws Exception 
	 */
	private void parseUnsubscribe(Map<String, Object> headers, String context, Json subscription) throws Exception {
		//get values
		String path = context+"."+subscription.at(PATH).asString();
		String wsSession = (String) headers.get(WebsocketConstants.CONNECTION_KEY);
		long period = 1000;
		if(subscription.at(PERIOD)!=null)period = subscription.at(PERIOD).asInteger();
		String format = FORMAT_DELTA;
		if(subscription.at(FORMAT)!=null)format=subscription.at(FORMAT).asString();
		String policy = POLICY_FIXED;
		if(subscription.at(POLICY)!=null)policy=subscription.at(POLICY).asString();
		long minPeriod = 0;
		if(subscription.at(MIN_PERIOD)!=null)minPeriod=subscription.at(MIN_PERIOD).asInteger();
		Subscription sub = new Subscription(wsSession, path, period, minPeriod, format, policy);
		SubscriptionManagerFactory.getInstance().removeSubscription(sub);
		
	}

	
} public class LoggerProcessor extends SignalkProcessor implements Processor {
	private static Logger logger = LogManager.getLogger(LoggerProcessor.class);

	@Override
	public void process(Exchange exchange) throws Exception {
		
		logger.debug("LoggerProcessor starts");
		HttpServletRequest request = exchange.getIn(HttpMessage.class).getRequest();
		logger.debug("Session = "+request.getSession().getId());
		HttpSession session = request.getSession();
		if (logger.isDebugEnabled()) {

			logger.debug("Request = " + exchange.getIn().getHeader(Exchange.HTTP_SERVLET_REQUEST).getClass());
			logger.debug("Session = " + session.getId());
		}

		if (session.getId() != null) {

			String remoteAddress = request.getRemoteAddr();
			String localAddress = request.getLocalAddr();
			if(Util.sameNetwork(localAddress, remoteAddress)){
				exchange.getIn().setHeader(SignalKConstants.MSG_TYPE, SignalKConstants.INTERNAL_IP);
			}else{
				exchange.getIn().setHeader(SignalKConstants.MSG_TYPE, SignalKConstants.EXTERNAL_IP);
			}

			if (exchange.getIn().getHeader(Exchange.HTTP_METHOD).equals("GET")) {
				processGet(exchange);
			}
			if (exchange.getIn().getHeader(Exchange.HTTP_METHOD).equals("POST")) {
				processPost(exchange);
			}
		} else {
			exchange.getIn().setHeader("Location", SignalKConstants.SIGNALK_AUTH);
			exchange.getIn().setBody("Authentication Required");
		}
	}

	private void processPost(Exchange exchange) throws IOException {
		String conf = exchange.getIn().getBody(String.class);
		//Json confJson = Json.read(conf);
		logger.debug("POST Log4j2 = " + conf);
		FileUtils.writeStringToFile(new File(Util.getRootPath()+"./conf/log4j2.json"), conf, StandardCharsets.UTF_8);

	}

	private void processGet(Exchange exchange) throws IOException {
		// get and return the current log4j2.json
		String conf = FileUtils.readFileToString(new File(Util.getRootPath()+"./conf/log4j2.json"));
		logger.debug("GET Log4j2 = " + conf);
		exchange.getIn().setBody(conf);
	}

} public class MapToJsonProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(MapToJsonProcessor.class);
	JsonSerializer ser = new JsonSerializer();
	
	public void process(Exchange exchange) throws Exception {
		
		if (exchange.getIn().getBody()==null)
			return;
		if(logger.isDebugEnabled())logger.debug("Processing:"+exchange.getIn().getBody().getClass());
		
		if (exchange.getIn().getBody() instanceof SignalKModel){
			SignalKModel model = (SignalKModel)exchange.getIn().getBody();
		
			exchange.getIn().setBody(ser.writeJson(model));
		}
		if(logger.isDebugEnabled())logger.debug("Outputting:"+exchange.getIn());
	}

} public class MqttProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(MqttProcessor.class);

	
	public void process(Exchange exchange) throws Exception {

		try {
			logger.debug("Message: "+exchange.getIn().getBody().getClass());
			logger.debug("Message: "+exchange.getIn().getBody());
			if (exchange.getIn().getBody() == null || !(exchange.getIn().getBody() instanceof String)){
				return;
			}
			String msg = exchange.getIn().getBody(String.class);
			exchange.getIn().setBody(msg.getBytes());
			String dest = exchange.getIn().getHeader(ConfigConstants.DESTINATION, String.class);
			if(StringUtils.isNotBlank(dest)){
				exchange.getIn().setHeader(ConfigConstants.CamelMQTTPublishTopic, dest.replace('.', '/'));
			}
			if(logger.isDebugEnabled()){
				logger.debug("Message: "+msg);
				logger.debug("Message destination :" + exchange.getIn().getHeader(ConfigConstants.CamelMQTTPublishTopic));
			}
			
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

} public class N2KProcessor extends SignalkProcessor implements Processor {

	
	private static Logger logger = LogManager.getLogger(N2KProcessor.class);
	
	private N2KHandler n2k = new N2KHandler(); 

	

	public void process(Exchange exchange) throws Exception {

		try {
			if (exchange.getIn().getHeader(N2K_MESSAGE) == null)
				return;
			if (exchange.getIn().getBody() == null || !(exchange.getIn().getBody() instanceof Json))
				return;

			SignalKModel model = n2k.handle(exchange.getIn().getHeader(N2K_MESSAGE, String.class));
			if(model!=null){
				if (logger.isDebugEnabled())
					logger.debug("Converted to:" + model);
				exchange.getIn().setBody(model);
			}
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}


} public class NMEA0183ExportProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(NMEA0183ExportProcessor.class);
	
	private NMEA0183Producer handler = new NMEA0183Producer();
	
	public void process(Exchange exchange) throws Exception {
		
		try {
			sendNmea(handler.createRMC(signalkModel));
			sendNmea(handler.createDBT(signalkModel));	
			sendNmea(handler.createGLL(signalkModel));		
			sendNmea(handler.createVTG(signalkModel));	
			sendNmea(handler.createHDG(signalkModel));	
			sendNmea(handler.createHDT(signalkModel));
			sendNmea(handler.createHDM(signalkModel));
			sendNmea(handler.createMWVApparent(signalkModel));	
			sendNmea(handler.createMWVTrue(signalkModel));	
			sendNmea(handler.createVHW(signalkModel));
			
		} catch (Exception e) {
			logger.error(e.getMessage());
			logger.debug(e.getMessage(),e);
			logger.debug("SignalkModel:"+signalkModel);
		}
	}

	private void sendNmea(String nmea) {
		if(logger.isDebugEnabled())logger.debug("Sending: "+nmea);
		if(nmea!=null){
			asyncSendBodyAndHeaders(nmeaProducer, nmea, WebsocketConstants.CONNECTION_KEY, WebsocketConstants.SEND_TO_ALL);
		}
		
	}

} public class NMEAProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(NMEAProcessor.class);
	

	private NMEAHandler nmea = new NMEAHandler();

	

	public void process(Exchange exchange) throws Exception {
		if (exchange.getIn().getBody() == null || !(exchange.getIn().getBody() instanceof String)) {
			return;
		}
		
		String body = exchange.getIn().getBody(String.class);
		if(logger.isDebugEnabled())
				logger.debug(body + "\n");
		String src = exchange.getIn().getHeader(MSG_SRC_BUS, String.class);
		SignalKModel model = nmea.handle(body, src);
		if(model!=null){
			exchange.getIn().setBody(model);
		}
	}

	

} public class OutputFilterProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(OutputFilterProcessor.class);
	JsonSerializer ser = new JsonSerializer();
	
	public void process(Exchange exchange) throws Exception {
		
		if (exchange.getIn().getBody()==null)
			return;
		if(logger.isDebugEnabled())logger.debug("Processing, class="+exchange.getIn().getBody().getClass());
		//TODO: add more filters here
		if(exchange.getIn().getBody() instanceof Json){
			Json json = (Json)exchange.getIn().getBody();
			//remove _arduino
			try{
				json.at(SignalKConstants.vessels).at(SignalKConstants.self).delAt("_arduino");
			}catch(NullPointerException npe){}
			//remove _config
			try{
				json.at(SignalKConstants.vessels).at(SignalKConstants.self).delAt("_config");
			}catch(NullPointerException npe){}
			
			exchange.getIn().setBody(json.toString());
		}
		if (exchange.getIn().getBody() instanceof SignalKModel){
			SignalKModel model = (SignalKModel)exchange.getIn().getBody();
			//remove _arduino
			try{
				model.getFullData().remove(SignalKConstants.vessels_dot_self_dot+"_arduino");
			}catch(NullPointerException npe){}
			//remove _config
			try{
				model.getFullData().remove(SignalKConstants.vessels_dot_self_dot+"_config");
			}catch(NullPointerException npe){}
			
			exchange.getIn().setBody(ser.write(model));
		}
		if(logger.isDebugEnabled()){
			logger.debug("Outputting:"+exchange.getIn().getHeaders());
			logger.debug("Outputting:"+exchange.getIn());
		}
	}

} public class PermissionsProcessor extends SignalkProcessor implements Processor {

	@Override
	public void process(Exchange exchange) throws Exception {
		if (exchange.getIn().getBody() instanceof SignalKModel) {
			Subject subject = exchange.getIn().getHeader(Exchange.AUTHENTICATION, Subject.class);
			if(subject==null)return;
			for(Principal p:subject.getPrincipals()){
				p.getName();
				
			}
		}

	}

} public class RestApiProcessor extends SignalkProcessor implements Processor {

	public static final String REST_REQUEST = "REST_REQUEST";
	//private static final String SLASH = "/";
	private static final String LIST = "list";
	public static final String REST_WILDCARD = "REST_WILDCARD";
	private static Logger logger = LogManager.getLogger(RestApiProcessor.class);

	public RestApiProcessor() throws IOException {

	}

	@Override
	public void process(Exchange exchange) throws Exception {
		// the Restlet request should be available if needed
		HttpServletRequest request = exchange.getIn(HttpMessage.class)
				.getRequest();
		HttpSession session = request.getSession();
		if (logger.isDebugEnabled()) {

			logger.debug("Request = "
					+ exchange.getIn().getHeader(Exchange.HTTP_SERVLET_REQUEST)
							.getClass());
			logger.debug("Session = " + session.getId());
		}

		if (session.getId() != null) {
			exchange.getIn().setHeader(REST_REQUEST, "true");
			String remoteAddress = request.getRemoteAddr();
			String localAddress = request.getLocalAddr();
			if(Util.sameNetwork(localAddress, remoteAddress)){
				exchange.getIn().setHeader(SignalKConstants.MSG_TYPE, SignalKConstants.INTERNAL_IP);
			}else{
				exchange.getIn().setHeader(SignalKConstants.MSG_TYPE, SignalKConstants.EXTERNAL_IP);
			}
			exchange.getIn().setHeader(SignalKConstants.MSG_SRC_IP, remoteAddress);
			exchange.getIn().setHeader(SignalKConstants.MSG_SRC_IP_PORT, request.getRemotePort());
			exchange.getIn().setHeader(SignalKConstants.MSG_SRC_BUS, "rest."+remoteAddress.replace('.', '_'));
			exchange.getIn().setHeader(WebsocketConstants.CONNECTION_KEY,
					session.getId());
			
			String path = (String) exchange.getIn()
					.getHeader(Exchange.HTTP_URI);
			if (logger.isDebugEnabled()) {
				logger.debug(exchange.getIn().getHeaders());
				logger.debug(path);
			}

			if (logger.isDebugEnabled())
				logger.debug("Processing the path = " + path);
			if (!isValidPath(path)) {
				exchange.getIn().setBody("Bad Request");
				exchange.getIn().setHeader(Exchange.CONTENT_TYPE, "text/plain");
				exchange.getIn().setHeader(Exchange.HTTP_RESPONSE_CODE,
						HttpServletResponse.SC_BAD_REQUEST);
				// response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
				return;
			}
			if (exchange.getIn().getHeader(Exchange.HTTP_METHOD).equals("GET")) {
				processGet(exchange, path);
			}
			if (exchange.getIn().getHeader(Exchange.HTTP_METHOD).equals("PUT")) {
				processPut(exchange, path);
			}
			if (exchange.getIn().getHeader(Exchange.HTTP_METHOD).equals("POST")) {
				if (exchange.getIn().getBody() instanceof StreamCache) {
					StreamCache cache = exchange.getIn().getBody(
							StreamCache.class);
					ByteArrayOutputStream writer = new ByteArrayOutputStream();
					cache.writeTo(writer);
					if (logger.isDebugEnabled())
						logger.debug("Reading the POST request:"+writer.toString());
					exchange.getIn().setBody(writer.toString());

					// POST here
					if (logger.isDebugEnabled())
						logger.debug("Processing the POST request:"
								+ exchange.getIn().getBody());
				}else{
					if (logger.isDebugEnabled())
						logger.debug("Skipping processing the POST request:"
								+ exchange.getIn().getBody().getClass());
				}
			}

		} else {
			// HttpServletResponse response =
			// exchange.getIn(HttpMessage.class).getResponse();
			exchange.getIn().setHeader(Exchange.HTTP_RESPONSE_CODE,
					HttpServletResponse.SC_MOVED_TEMPORARILY);
			// constant("http://somewhere.com"))
			exchange.getIn().setHeader("Location", SignalKConstants.SIGNALK_AUTH);
			exchange.getIn().setBody("Authentication Required");
		}

	}

	private void processPut(Exchange exchange, String path) {
		path = standardizePath(path);

		String context = Util.getContext(path);
		if (logger.isDebugEnabled())
			logger.debug("Processing the PUT context:" + context);
		if (path.length() > context.length()) {
			path = path.substring(context.length() + 1);
		}
		// make PUT object
		// "{\"context\":\"vessels.*\",\"put\":[{"values":{\"path\":\"navigation.courseOverGroundTrue\", "value": 172.3}, "source": {"device": "/dev/actisense", "timestamp": "2014-08-15-16:00:00.081","src": "115", "pgn": "128267" }]}";
		exchange.getIn().setHeader(Exchange.CONTENT_TYPE, "application/json");
		Json json = Json.object().set(SignalKConstants.CONTEXT, context);
		Json array = Json.array();
		json.set(SignalKConstants.PUT, array);
		Json entry = Json.object();
		// add the source
		entry.set(SignalKConstants.sourceRef, Json.object());
		// add the value
		Json values = Json.array();
		entry.set(SignalKConstants.values, values);
		values.set(SignalKConstants.PATH, path);
		values.set(SignalKConstants.value,
				Json.read(exchange.getIn().getBody(String.class)));

		exchange.getIn().setBody(json.toString());

		if (logger.isDebugEnabled())
			logger.debug("Processing the PUT request:"
					+ exchange.getIn().getBody());

	}

	private boolean isValidPath(String path) {
		if (StringUtils.isBlank(path))
			return false;
		if (path.equals(SignalKConstants.SIGNALK_DISCOVERY))
			return true;
		if (path.startsWith(SignalKConstants.SIGNALK_API))
			return true;
		if (path.startsWith(SignalKConstants.SIGNALK_CONFIG))
			return true;
		return false;
	}

	private void processGet(Exchange exchange, String path) throws UnknownHostException {

		// discovery request
		if (path.equals(SignalKConstants.SIGNALK_DISCOVERY)) {
			doDiscovery(exchange, path);
			return;
		}
		
		//self
		if (path.equals(SignalKConstants.SIGNALK_API+"/self")) {
			path=SignalKConstants.SIGNALK_API+"/vessels/self/uuid";
			exchange.getIn().setHeader(Exchange.HTTP_PATH,"/vessels/self/uuid");
		}

		path = standardizePath(path);

		
		String context = Util.getContext(path);
		if (logger.isDebugEnabled())
			logger.debug("Processing the context:" + context);
		if (path.length() > context.length() && context.length()>0) {
			path = path.substring(context.length() + 1);
		} else {
			path = "*";
		}
		// list
		if (context.startsWith(LIST)) {
			// make LIST obj
			doList(exchange, context, path);
			return;
		}
		// make GET obj
		// "{\"context\":\"vessels.*\",\"get\":[{\"path\":\"navigation.*\"}]}";
		exchange.getIn().setHeader(Exchange.CONTENT_TYPE, "application/json");
		Json json = Json.object().set(SignalKConstants.CONTEXT, context);
		Json array = Json.array().add(
				Json.object().set(SignalKConstants.PATH, path));
		
		json.set(SignalKConstants.GET, array);
		exchange.getIn().setBody(json.toString());
		// If a GET is an absolute object return only the requested object
		// If its a wildcard, return a full tree
		if (containsWildcard(context) || containsWildcard(path)) {
			exchange.getIn().setHeader(REST_WILDCARD, "true");
		} else {
			exchange.getIn().setHeader(REST_WILDCARD, "false");
		}
		if (logger.isDebugEnabled())
			logger.debug("Processing the GET request:"
					+ exchange.getIn().getBody());

	}

	private void doList(Exchange exchange, String context, String path) {
		// "{\"context\":\"vessels.*\",\"list\":[{\"path\":\"navigation.*\"}]}";
		exchange.getIn().setHeader(Exchange.CONTENT_TYPE,
				"application/json");
		Json json = Json.object().set(SignalKConstants.CONTEXT,
				context.substring(LIST.length() + 1));
		Json array = Json.array().add(
				Json.object().set(SignalKConstants.PATH, path));
		json.set(SignalKConstants.LIST, array);
		exchange.getIn().setBody(json.toString());
		if (logger.isDebugEnabled())
			logger.debug("Processing the LIST request:"
					+ exchange.getIn().getBody());
		
	}

	private void doDiscovery(Exchange exchange, String path) {
		Message in = exchange.getIn();

		String hostname = Util.getConfigProperty(ConfigConstants.HOSTNAME);
		if (StringUtils.isBlank(hostname)) {
			try {
				String header = (String) in.getHeader(Exchange.HTTP_URL);
				hostname = new URI(header).getHost();
			} catch (URISyntaxException e) {
				// Should not happen as we expect Camel to return a valid URI.
				logger.warn("Invalid URI returned from Exchange: " + in.getHeader(Exchange.HTTP_URL));
				hostname = "localhost";
			}
		}

		in.setHeader(Exchange.CONTENT_TYPE, "application/json");
		in.setHeader(Exchange.HTTP_RESPONSE_CODE, HttpServletResponse.SC_OK);
		in.setBody(discovery(hostname).toString());
		
	}

	// TODO: This should come from the configuration used to start the endpoints.
	static Json discovery(String hostname) {
		
		Json version = Json.object();
		String ver = Util.getConfigProperty(ConfigConstants.VERSION);
		if( ver.startsWith("v")){
			ver = ver.substring(1);
		}
		version.set("version", ver);
		version.set(SignalKConstants.websocketUrl, "ws://" + hostname + ":"
				+ Util.getConfigPropertyInt(ConfigConstants.WEBSOCKET_PORT)
				+ SignalKConstants.SIGNALK_WS);
		version.set(SignalKConstants.restUrl, "http://" + hostname + ":"
				+ Util.getConfigPropertyInt(ConfigConstants.REST_PORT)
				+ SignalKConstants.SIGNALK_API + "/");
		version.set(SignalKConstants.signalkTcpPort, "tcp://" + hostname + ":"
				+ Util.getConfigPropertyInt(ConfigConstants.TCP_PORT));
		version.set(SignalKConstants.signalkUdpPort, "udp://" + hostname + ":"
				+ Util.getConfigPropertyInt(ConfigConstants.UDP_PORT));
		version.set(SignalKConstants.nmeaTcpPort, "tcp://" + hostname + ":"
				+ Util.getConfigPropertyInt(ConfigConstants.TCP_NMEA_PORT));
		version.set(SignalKConstants.nmeaUdpPort, "udp://" + hostname + ":"
				+ Util.getConfigPropertyInt(ConfigConstants.UDP_NMEA_PORT));
		if (Util.getConfigPropertyBoolean(ConfigConstants.START_STOMP))
			version.set(SignalKConstants.stompPort, "stomp+nio://" + hostname + ":"
					+ Util.getConfigPropertyInt(ConfigConstants.STOMP_PORT));
		if (Util.getConfigPropertyBoolean(ConfigConstants.START_MQTT))
			version.set(SignalKConstants.mqttPort, "mqtt://" + hostname + ":"
					+ Util.getConfigPropertyInt(ConfigConstants.MQTT_PORT));
		Json endpoints = Json.object();
		endpoints.set("v"+ver.substring(0, ver.indexOf(".")), version );
		return Json.object().set("endpoints", endpoints);
	}

	private String standardizePath(String path) {

		// check valid request.

		path = path.substring(SignalKConstants.SIGNALK_API.length());
		if (path.startsWith("/"))
			path = path.substring(1);
		if (path.endsWith("/"))
			path = path.substring(0, path.length() - 1);

		path = path.replace("/", ".");
		if (logger.isDebugEnabled())
			logger.debug("Processing the path extension:" + path);
		return path;
	}

	/**
	 * true if the path contains any * or ? for a wildcard match
	 * 
	 * @param path
	 * @return
	 */
	private boolean containsWildcard(String path) {
		if (StringUtils.isBlank(path))
			return false;
		if (path.contains("*") || path.contains("?"))
			return true;
		return false;
	}

} public class RestAuthProcessor extends SignalkProcessor implements Processor{

	private static Logger logger = LogManager.getLogger(RestAuthProcessor.class);
	
	@Override
	public void process(Exchange exchange) throws Exception {
		// the Restlet request should be available if neeeded
		HttpServletRequest request = exchange.getIn(HttpMessage.class).getRequest();
		
		 //HttpSession session = request.getSession();
		if(logger.isDebugEnabled())logger.debug("Session = "+request.getSession().getId());
       // Request request = exchange.getIn().getHeader(RestletConstants.RESTLET_REQUEST, Request.class);
        if("GET"==request.getMethod()){
        	processGet(request, exchange);
        }else{
        	exchange.getIn(HttpMessage.class).getResponse().setStatus(HttpServletResponse.SC_BAD_REQUEST);
        }
	}

	private void processGet(HttpServletRequest request, Exchange exchange) throws Exception {
		// use Restlet API to create the response
		
        HttpServletResponse response = exchange.getIn(HttpMessage.class).getResponse();
        String path =  exchange.getIn().getHeader(Exchange.HTTP_URI, String.class);
        if(logger.isDebugEnabled())logger.debug("We are processing the path = "+path);
        
        //check valid request.
        if(path.length()<=SignalKConstants.SIGNALK_AUTH.length()){
        	response.setStatus(HttpServletResponse.SC_BAD_REQUEST);
        	return;
        }
       // path=path.substring(request.getRootRef().getPath().length());
      
        //logger.debug("We are processing the extension:"+path);
       //TODO: sort out proper security here!
      
        response.setContentType(MimeTypes.TEXT_PLAIN );
        //String uuid = UUID.randomUUID().toString();
        //Cookie cookie = new Cookie(Constants.SESSIONID, uuid);
       // cookie.setPath("/signalk/");
       // response.addCookie(cookie);
        //String breadcrumb = exchange.getIn().getHeader(Exchange.BREADCRUMB_ID,String.class);
        //breadcrumb = breadcrumb.substring(0,breadcrumb.lastIndexOf("-",breadcrumb.lastIndexOf("-")));
        //manager.add(cookieSetting.getValue(), cookieSetting.getValue());
        if(logger.isDebugEnabled())logger.info("Adding session = "+request.getSession().getId());
        //manager.add(request.getSession().getId(), request.getSession().getId(), Constants.OUTPUT_WS);
       
        // SEND RESPONSE
        //exchange.getOut().setBody(response.getEntityAsText());
       
        response.setStatus(HttpServletResponse.SC_OK);
        //send back
        //response.redirectSeeOther(request.getReferrerRef());
		
	}

} public class RestPathFilterProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(RestPathFilterProcessor.class);
	JsonSerializer ser = new JsonSerializer();
	
	
	public void process(Exchange exchange) throws Exception {
		
		if (exchange.getIn().getBody()==null)
			return;
		if(logger.isDebugEnabled())logger.debug("Processing, class="+exchange.getIn().getBody().getClass());
		//TODO: add more filters here
		Json reply = null;
		if (exchange.getIn().getBody() instanceof String){			
			reply = Json.read(exchange.getIn().getBody(String.class));
		}else if (exchange.getIn().getBody() instanceof SignalKModel){
			SignalKModel model = (SignalKModel)exchange.getIn().getBody();
			reply = ser.writeJson(model);
		}else if(exchange.getIn().getBody() instanceof Json){
			reply = (Json)exchange.getIn().getBody();
			
		}
		//trim leading path
		String path = exchange.getIn().getHeader(Exchange.HTTP_PATH, String.class);
		//not for /list POST PUT
		if(!exchange.getIn().getHeader(Exchange.HTTP_METHOD).equals("GET") || path.startsWith("/"+LIST))
			return;
		path = sanitizePath(path);
		if(path.endsWith("*")){
			path=path.substring(0, path.lastIndexOf("."));
		}
		
		if(logger.isDebugEnabled()){
			logger.debug("Trimming:"+exchange.getIn().getHeaders());
			logger.debug("Trimming by "+path+" : "+reply);
		}
		if(StringUtils.isNotBlank(path)&& reply !=null && !reply.toString().equals("{}")){
			reply = Util.findNode(reply, path); 
			
		}
		if(reply==null){
			exchange.getIn().setBody("Bad Request");
			exchange.getIn().setHeader(Exchange.CONTENT_TYPE, "text/plain");
			exchange.getIn().setHeader(Exchange.HTTP_RESPONSE_CODE,
					HttpServletResponse.SC_BAD_REQUEST);
			return;
		}
		//all ok
		exchange.getIn().setBody(reply.toString());
		
		
		if(logger.isDebugEnabled()){
			logger.debug("Outputting:"+exchange.getIn().getHeaders());
			logger.debug("Outputting:"+exchange.getIn());
		}
	}

} public class SaveProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(SaveProcessor.class);
	private static Logger history = LogManager.getLogger("nz.co.fortytwo.signalk.history");
	
	private SaveThread saver = new SaveThread();
	
	public SaveProcessor(){
		signalkModel.getEventBus().register(this);
	}
	
	public void process(Exchange exchange) throws Exception {
		//do nothing, event driven
		
	}

	@Subscribe
	public void recordEvent(PathEvent pathEvent) {
		if (pathEvent == null)
			return;
		if (pathEvent.getPath() == null)
			return;
		//if (logger.isTraceEnabled())logger.trace(this.wsSession + " received event " + pathEvent.getPath());
		history.info(String.format("%d|%s|%s", pathEvent.getTimestamp(), pathEvent.getPath(), signalkModel.get(pathEvent.getPath())));
		// do we care?
		saver.startSave();

	}

	@Subscribe
	public void recordEvent(DeadEvent e) {
		logger.debug("Received dead event" + e.getSource());
	}

	
	
	class SaveThread implements Runnable {
		long lastSave = 0;

		Thread t = null;

		public void startSave() {
			if(logger.isDebugEnabled())logger.debug("Checking save..");
			if (t != null && t.isAlive())
				return;
			if(logger.isDebugEnabled())logger.debug("Starting save..");
			t = new Thread(this);
			t.start();
		}

		@Override
		public void run() {
			
			while (true) {
				if (System.currentTimeMillis() - lastSave > 60000) {
					try {
						SignalKModelFactory.save(signalkModel);
						lastSave = System.currentTimeMillis();
					} catch (IOException e) {
						logger.error(e.getMessage());
						logger.debug(e);
					}
					
					break;
				} else {
					try {
						Thread.sleep(500);
					} catch (InterruptedException e) {
					}
				}
			}

		}

		

	}
} public class SignalkModelProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(SignalkModelProcessor.class);

	public void process(Exchange exchange) throws Exception {

		try {

			if (exchange.getIn().getBody() instanceof SignalKModel) {
				handle(exchange.getIn().getBody(SignalKModel.class));
				//we want to stop here
				exchange.getIn().setBody(null);
			}else{
				if(logger.isDebugEnabled())logger.debug("Ignored, not update:"+exchange.getIn().getBody(Json.class));
			}

		} catch (IllegalArgumentException e) {
			logger.info(e.getMessage());
		}catch (Exception e) {
				logger.error(e.getMessage(), e);
		}
	}

	// @Override
	public void handle(SignalKModel node) throws IOException {
		if (node.getFullData().size() == 0)
			return;
		if (logger.isDebugEnabled())
            logger.debug("SignalkModelProcessor  updating " + SignalKModelFactory.writePretty(node));

		signalkModel.putAll(node.getFullData());
		NavigableMap<String, Object> cnf = node.getSubMap(SignalKConstants.CONFIG);
		if(cnf!=null && cnf.size()>0){
			SignalKModelFactory.saveConfig(signalkModel);
		}

	}

} public class SourceRefToSourceProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(SourceRefToSourceProcessor.class);
	JsonSerializer ser = new JsonSerializer();
	
	public void process(Exchange exchange) throws Exception {
		
		if (exchange.getIn().getBody()==null)
			return;
		if(logger.isDebugEnabled())logger.debug("Processing:"+exchange.getIn().getBody().getClass());
		
		if (exchange.getIn().getBody() instanceof SignalKModel){
			SignalKModel model = exchange.getIn().getBody(SignalKModel.class);
			if(logger.isDebugEnabled())logger.debug("Processing:"+model);
			for(String key : model.getKeys()){
				if(key.endsWith(dot+sourceRef)){
					String val = (String) model.get(key);
					if(logger.isDebugEnabled())logger.debug("key:"+key+", value:"+val);
					//val is the path
					model.getFullData().remove(key);
					//fix key
					key = key.replace(dot+sourceRef, dot+source);
					NavigableMap<String, Object> node = signalkModel.getSubMap(sources+dot+val);
					int pos = (sources+dot+val).length();
					//node is the source object 
					if(node!=null){
						for(Entry<String, Object> entry:node.entrySet()){
							String nodeKey = entry.getKey();
							model.getFullData().put(key+nodeKey.substring(pos), entry.getValue());
						}
					}
				}
			}
			exchange.getIn().setBody(model);
		}
		if(logger.isDebugEnabled())logger.debug("Outputting:"+exchange.getIn());
	}

} public class SourceToSourceRefProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(SourceToSourceRefProcessor.class);
	
	
	public void process(Exchange exchange) throws Exception {
		
		if (exchange.getIn().getBody()==null)
			return;
		if(logger.isDebugEnabled())logger.debug("Processing:"+exchange.getIn().getBody().getClass());
		
		if (exchange.getIn().getBody() instanceof SignalKModel){
			SignalKModel model = exchange.getIn().getBody(SignalKModel.class);
			if(logger.isDebugEnabled())logger.debug("Processing:"+model);
			for(String key : model.getKeys()){
				//get the source.type key
				//if(logger.isDebugEnabled())logger.debug("Key:"+key);
				if(key.endsWith(dot+source+dot+label)){
					if(logger.isDebugEnabled())logger.debug("Convert key:"+key);
					int pos = key.indexOf(dot+source+dot);
					String typeVal = exchange.getIn().getHeader(MSG_TYPE,String.class);
					
					int pos1 = pos+source.length()+2;
					String refKey = key.substring(0,pos);
					if(logger.isDebugEnabled())logger.debug("refKey:"+refKey+", bus:"+typeVal);
					//get the label
					String lbl = (String) model.get(refKey+dot+source+dot+label);
					if(logger.isDebugEnabled())logger.debug("refKey:"+refKey+", label:"+lbl);
					//set sourceRef
					model.getFullData().put(refKey+dot+sourceRef, typeVal+dot+lbl);
					//put in sources
					NavigableMap<String, Object> node = signalkModel.getSubMap(refKey+dot+source);
					if(logger.isDebugEnabled())logger.debug("Found keys:"+node.size());
					//node is the source object 
					if(node!=null){
						for(Entry<String, Object> entry:node.entrySet()){
							String nodeKey = entry.getKey().substring(pos1);
							model.getFullData().put(sources+dot+typeVal+dot+lbl+dot+nodeKey, entry.getValue());
							if(logger.isDebugEnabled())logger.debug("Added key:"+sources+dot+typeVal+dot+lbl+dot+nodeKey+"="+entry.getValue());
						}
					
						//drop the source key and all subkeys
						for(Entry<String, Object> entry:node.entrySet()){
							String nodeKey = entry.getKey();
							model.getFullData().remove(nodeKey);
							if(logger.isDebugEnabled())logger.debug("Remove key:"+nodeKey);
						}
					}
					
				}
			}
			exchange.getIn().setBody(model);
			if(logger.isDebugEnabled())logger.debug("Outputting:"+exchange.getIn());
		}
		
		
	}



} public class StompProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(StompProcessor.class);
	
	public void process(Exchange exchange) throws Exception {

		try {
			if (exchange.getIn().getBody() == null || !(exchange.getIn().getBody() instanceof Json))
				return;
			
			if(logger.isDebugEnabled())logger.debug("Message destination :" + exchange.getIn().getHeader(Constants.DESTINATION.toString()));
			
			exchange.getIn().setHeader("ttl", "3000");
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

} public class TrackProcessor extends SignalkProcessor implements Processor {

	private static final String COORDINATES = "coordinates";
	private static final String GEOMETRY = "geometry";
	private static final String FEATURE = "feature";
	private static final String GEOJSON = "[{\"geometry\":{\"coordinates\":[],\"type\":\"LineString\"},\"properties\":null,\"id\":\"na\",\"type\":\"Feature\"}]";

	// simplify to about 2m out of true (at equator)
	private static final double TRACK_TOLERANCE = 0.00002;

	private static final int MAX_COUNT = 5000;
	private static final int SAVE_COUNT = 60;

	private static Logger logger = LogManager.getLogger(TrackProcessor.class);
	
	private Json msg ;
	//TODO: thread safety??
	private List<Position> track = new ArrayList<Position>();
	private Json coords;
	private Json geometry;
	private static int count = 0;
	private boolean initialized = false;

	public TrackProcessor() throws Exception {
		msg = createTrackMsg();
	}
	

	public void process(Exchange exchange) throws Exception {
		try {
			if(logger.isDebugEnabled()&& exchange.getIn().getBody()!=null){
				logger.debug("Processing:"+exchange.getIn().getBody().getClass());
				logger.trace("Processing:"+exchange.getIn().getBody());
			}
			if(!initialized){
				//save new message
				inProducer.sendBody(msg.toString());
				initialized=true;
			}
			if (exchange.getIn().getBody() instanceof SignalKModel) {

				handle(exchange.getIn().getBody(SignalKModel.class));
			}else{
				if(logger.isTraceEnabled())logger.trace("Ignored, not a track:"+exchange.getIn().getBody(Json.class));
			}
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

	// @Override
	public void handle(SignalKModel node) {
		if (node.getData().size() == 0)
			return;

		if (node.getData().containsKey(vessels_dot_self_dot + nav_position_latitude) && node.getData().containsKey(vessels_dot_self_dot + nav_position_longitude)) {
			if (logger.isTraceEnabled())
				logger.trace("TrackProcessor  updating " + node);
			// we have a track change.
			track.add(new Position((double)node.get(vessels_dot_self_dot + nav_position_latitude), (double)node.get(vessels_dot_self_dot + nav_position_longitude)));
			count++;
			// append to file
			if (count % SAVE_COUNT == 0) {
				// save it
				for(Position p: track){
					coords.add(Json.array(p.longitude(), p.latitude()));
				}
				geometry.set(COORDINATES, coords);
				if(logger.isDebugEnabled())logger.debug("Saving track:"+count);
				updateSourceAndTime(msg);
				inProducer.asyncSendBody(inProducer.getDefaultEndpoint(), msg.toString());
				coords.asJsonList().clear();
			}
			if (count % (SAVE_COUNT * 4) == 0) {
				if (logger.isDebugEnabled())
					logger.debug("Simplify track, size:" + coords.asList().size());// +":"+coords);
				track = TrackSimplifier.simplify(track, TRACK_TOLERANCE);

				if (logger.isDebugEnabled())
					logger.debug("Simplify track, done, size:" + coords.asList().size());
				count = track.size();
			}
			// reset?
			if (count > MAX_COUNT) {
				if(logger.isDebugEnabled())logger.debug("Rolling to new track:"+count);
				for(Position p: track){
					coords.add(Json.array(p.longitude(), p.latitude()));
				}
				geometry.set(COORDINATES, coords);
				updateSourceAndTime(msg);
				inProducer.asyncSendBody(inProducer.getDefaultEndpoint(), msg.toString());
				msg = createTrackMsg();
				//save new message
				inProducer.asyncSendBody(inProducer.getDefaultEndpoint(), msg.toString());
				count = 0;
				track.clear();
			}
		}
	}


	private Json createTrackMsg(){
		Json val = Json.object();
		val.set(SignalKConstants.PATH, routes + dot + "urn:mrn:signalk:uuid:"+UUID.randomUUID().toString());
		Json currentTrack = Json.object();
		val.set(value, currentTrack);
		String time = Util.getIsoTimeString();
		time = time.substring(0, time.indexOf("."));
		currentTrack.set(name, "Vessel Track from "+time);
		currentTrack.set("description", "Auto saved vessel track from "+time);
		
		Json values = Json.array();
		values.add(val);

		Json update = Json.object();
		
		update.set(SignalKConstants.values, values);

		Json updates = Json.array();
		updates.add(update);
		Json msg = Json.object();
		msg.set(SignalKConstants.CONTEXT, resources);
		msg.set(SignalKConstants.PUT, updates);
		updateSourceAndTime(msg);
		Json geoJson = Json.read(GEOJSON);
		geometry = geoJson.at(0).at(GEOMETRY);
		coords = geometry.at(COORDINATES);
		currentTrack.set(FEATURE, geoJson);
		if(logger.isDebugEnabled())logger.debug("Created new track msg:"+msg);
		return msg;
	}


	private void updateSourceAndTime(Json msg) {
		Json update = msg.at(PUT).at(0);
		update.set(timestamp, Util.getIsoTimeString());
		update.set(source, Json.read("{\"internal\": {"+
	          "\"value\": \"TrackProcessor\","+
	          "\"timestamp\": \""+Util.getIsoTimeString()+"\"}}"));
	}
} public class UploadProcessor extends SignalkProcessor implements Processor {
	private static Logger logger = LogManager.getLogger(UploadProcessor.class);

	@Override
	public void process(Exchange exchange) throws Exception {

		logger.debug("UploadProcessor starts");
		HttpServletRequest request = exchange.getIn(HttpMessage.class).getRequest();
		logger.debug("Session = " + request.getSession().getId());
		HttpSession session = request.getSession();
		if (logger.isDebugEnabled()) {

			logger.debug("Request = " + exchange.getIn().getHeader(Exchange.HTTP_SERVLET_REQUEST).getClass());
			logger.debug("Session = " + session.getId());
		}

		if (session.getId() != null) {

			String remoteAddress = request.getRemoteAddr();
			String localAddress = request.getLocalAddr();
			if (Util.sameNetwork(localAddress, remoteAddress)) {
				exchange.getIn().setHeader(SignalKConstants.MSG_TYPE, SignalKConstants.INTERNAL_IP);
			} else {
				exchange.getIn().setHeader(SignalKConstants.MSG_TYPE, SignalKConstants.EXTERNAL_IP);
			}
			if (exchange.getIn().getHeader(Exchange.HTTP_METHOD).equals("POST")) {
				 processUpload(exchange);
			}
		} else {
			exchange.getIn().setHeader("Location", SignalKConstants.SIGNALK_AUTH);
			exchange.getIn().setBody("Authentication Required");
		}
	}

	private void processUpload(Exchange exchange) throws Exception {
		logger.debug("Begin import:"+ exchange.getIn().getHeaders());
		if(exchange.getIn().getBody()!=null){
			logger.debug("Body class:"+ exchange.getIn().getBody().getClass());
		}else{
			logger.debug("Body class is null");
		}
		//logger.debug("Begin import:"+ exchange.getIn().getBody());
		MediaType mediaType = exchange.getIn().getHeader(Exchange.CONTENT_TYPE, MediaType.class);
		InputRepresentation representation = new InputRepresentation((InputStream) exchange.getIn().getBody(), mediaType);
		logger.debug("Found MIME:"+ mediaType+", length:"+exchange.getIn().getHeader(Exchange.CONTENT_LENGTH, Integer.class));
		//make a reply
		Json reply = Json.read("{\"files\": []}");
		Json files = reply.at("files");
		
		try {
			List<FileItem> items = new RestletFileUpload(new DiskFileItemFactory()).parseRepresentation(representation);
			logger.debug("Begin import files:"+items);
			for (FileItem item : items) {
				if (!item.isFormField()) {
					InputStream inputStream = item.getInputStream();
					
					Path destination = Paths.get(Util.getConfigProperty(STATIC_DIR)+Util.getConfigProperty(MAP_DIR)+item.getName());
					logger.debug("Save import file:"+destination);
					long len = Files.copy(inputStream, destination, StandardCopyOption.REPLACE_EXISTING);
					Json f = Json.object();
					f.set("name",item.getName());
					f.set("size",len);
					files.add(f);
					install(destination);
				}
			}
		} catch (FileUploadException | IOException e) {
			logger.error(e.getMessage(),e);
		}
		exchange.getIn().setBody(reply);
	}

	private void install(Path destination) throws Exception {
		if(!destination.toString().endsWith(".zip"))return;
		//unzip here
		logger.debug("Unzipping file:"+destination);
		try{
			File zipFile = destination.toFile();
			String f = destination.toFile().getName();
			f= f.substring(0,f.indexOf("."));
			File destDir = new File(Util.getConfigProperty(STATIC_DIR)+Util.getConfigProperty(MAP_DIR)+f);
			if(!destDir.exists()){
				destDir.mkdirs();
			}
			ZipUtils.unzip(destDir, zipFile);
			logger.debug("Unzipped file:"+destDir);
			//now add a reference in resources
			loadChart(f);
		}catch(Exception e){
			logger.error(e.getMessage(),e);
			throw e;
		}
	}
	
	public void loadChart(String chartName) throws Exception{
		try{
			 File destDir = new File(Util.getConfigProperty(STATIC_DIR)+Util.getConfigProperty(MAP_DIR)+chartName);
			 SAXReader reader = new SAXReader();
		     Document document = reader.read(new File(destDir, "tilemapresource.xml"));
		     
		     String title = document.getRootElement().element("Title").getText();
		     String scale = "250000";
		     if(document.getRootElement().element("Metadata")!=null){
		    	 scale = document.getRootElement().element("Metadata").attribute("scale").getText();
		     }
		     double maxRes = 0.0;
		     double minRes = Double.MAX_VALUE;
		     int maxZoom = 0;
		     int minZoom = 99;
		     Element tileSets = document.getRootElement().element("TileSets");
		     for(Object o: tileSets.elements("TileSet")){
		    	 Element e = (Element)o;
		    	 int href = Integer.parseInt(e.attribute("href").getValue());
		    	 maxZoom=Math.max(href, maxZoom);
		    	 minZoom=Math.min(href, minZoom);
		    	 double units = Double.parseDouble(e.attribute("units-per-pixel").getValue());
		    	 maxRes=Math.max(units, maxRes);
		    	 minRes=Math.min(units, minRes);
		     }
		     //now make an entry in resources
		     Json resource = createChartMsg(chartName, title, scale);
		     inProducer.asyncSendBody(inProducer.getDefaultEndpoint(),resource);
		}catch(Exception e){
			logger.error(e.getMessage(),e);
			throw e;
		}
	}
	
	private Json createChartMsg(String f, String title, String scale){
		Json val = Json.object();
		val.set(SignalKConstants.PATH, "charts." + "urn:mrn:signalk:uuid:"+UUID.randomUUID().toString());
		Json currentChart = Json.object();
		val.set(value, currentChart);
		String time = Util.getIsoTimeString();
		time = time.substring(0, time.indexOf("."));
		currentChart.set("identifier", f);
		currentChart.set(name, title);
		currentChart.set("description", title);
		currentChart.set("tilemapUrl", "/"+Util.getConfigProperty(MAP_DIR)+f);
		try{
			int scaleInt = Integer.valueOf(scale);
			currentChart.set("scale", scaleInt);
		}catch(Exception e){
			currentChart.set("scale", 0);
		}
		
		Json values = Json.array();
		values.add(val);

		Json update = Json.object();
		
		update.set(SignalKConstants.values, values);

		Json updates = Json.array();
		updates.add(update);
		Json msg = Json.object();
		msg.set(SignalKConstants.CONTEXT, resources);
		msg.set(SignalKConstants.PUT, updates);
		
		
		if(logger.isDebugEnabled())logger.debug("Created new chart msg:"+msg);
		return msg;
	}

} public class ValidationProcessor extends SignalkProcessor implements Processor{

	private static Logger logger = LogManager.getLogger(ValidationProcessor.class);
	
	public void process(Exchange exchange) throws Exception {
		
		try {
			//if we have a REST_REPLY send it now
			
			if(exchange.getIn().getBody() instanceof SignalKModel){
				validate(exchange.getIn().getBody(SignalKModel.class));
			}else{
				if(logger.isDebugEnabled())logger.debug("Ignored, not update:"+exchange.getIn().getBody(Json.class));
			}
			
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
		}
	}

	
	/**
	 * @param model
	 */
	public void validate(SignalKModel model){
		//is this a leaf?
		if(logger.isDebugEnabled())	logger.debug(model.toString());
		if(model==null||model.getKeys().size()==0)return;
		//check the values
		for(String key: model.getKeys()){
			if(logger.isDebugEnabled())	logger.debug("Checking key="+key);
			if(key.endsWith(dot+value)){
				if(logger.isDebugEnabled())	logger.debug("Processing key="+key);
				String tmpKey = key.substring(0,key.length()-value.length());
				//it should have timestamp and source
				if(model.getSubMap(tmpKey+timestamp).size()==0){
					model.getFullData().put(tmpKey+timestamp,Util.getIsoTimeString());
				}
				if(model.getSubMap(tmpKey+sourceRef).size()==0){
					model.getFullData().put(tmpKey+sourceRef,UNKNOWN);
				}
			}
		
		}
		
	}

} public class WindProcessor extends SignalkProcessor implements Processor{

	private static Logger logger = LogManager.getLogger(WindProcessor.class);
	private TrueWindHandler trueWind = new TrueWindHandler();

	public void process(Exchange exchange) throws Exception {
		
		try {
		
			trueWind.handle(signalkModel);
			
		} catch (Exception e) {
			logger.error(e);
		}
	}
	

} public class WsSessionProcessor extends SignalkProcessor implements Processor {

	private static Logger logger = LogManager.getLogger(WsSessionProcessor.class);

	@Override
	public void process(Exchange exchange) throws Exception {
		// check the client
		// exchange.getIn().setHeader(SignalKConstants.MSG_TYPE,
		// SignalKConstants.INTERNAL_IP);
		String wsSession = exchange.getIn().getHeader(
				WebsocketConstants.CONNECTION_KEY, String.class);
		if (logger.isDebugEnabled()) {
			logger.debug("WS connection key is " + wsSession);
			for (String key : exchange.getIn().getHeaders().keySet()) {
				logger.debug("In headers = " + key + "="
						+ exchange.getIn().getHeader(key));
			}
			for (String key : exchange.getProperties().keySet()) {
				logger.debug("props = " + key);
			}
			logger.debug("Found sessionId session = " + wsSession + ","
					+ manager.getSessionId(wsSession));
		}
		// need to set the ipaddress header here
		String remoteAddress = manager.getRemoteIpAddress(wsSession);
		String localAddress = manager.getLocalIpAddress(wsSession);

		exchange.getIn().setHeader(SignalKConstants.MSG_SRC_IP, remoteAddress);
		exchange.getIn().setHeader(SignalKConstants.MSG_SRC_BUS, "ws."+remoteAddress.replace('.', '_'));
		
		if (Util.sameNetwork(localAddress, remoteAddress)) {
			exchange.getIn().setHeader(SignalKConstants.MSG_TYPE,
					SignalKConstants.INTERNAL_IP);
		} else {
			exchange.getIn().setHeader(SignalKConstants.MSG_TYPE,
					SignalKConstants.EXTERNAL_IP);
		}

	}

}
import java.io.File;
import java.io.IOException;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import javax.servlet.http.HttpServletResponse;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
import org.apache.camel.Predicate;
import org.apache.camel.Processor;
import org.apache.camel.builder.PredicateBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.websocket.WebsocketConstants;
import org.apache.camel.component.websocket.WebsocketEndpoint;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.model.RouteDefinition;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import mjson.Json;
import nz.co.fortytwo.signalk.processor.AISProcessor;
import nz.co.fortytwo.signalk.processor.AisExpiryProcessor;
import nz.co.fortytwo.signalk.processor.AlarmProcessor;
import nz.co.fortytwo.signalk.processor.AnchorWatchProcessor;
import nz.co.fortytwo.signalk.processor.ClientAppProcessor;
import nz.co.fortytwo.signalk.processor.ConfigFilterProcessor;
import nz.co.fortytwo.signalk.processor.DeclinationProcessor;
import nz.co.fortytwo.signalk.processor.DeltaImportProcessor;
import nz.co.fortytwo.signalk.processor.DepthProcessor;
import nz.co.fortytwo.signalk.processor.FullExportProcessor;
import nz.co.fortytwo.signalk.processor.FullImportProcessor;
import nz.co.fortytwo.signalk.processor.FullToDeltaProcessor;
import nz.co.fortytwo.signalk.processor.HeartbeatProcessor;
import nz.co.fortytwo.signalk.processor.IncomingSecurityFirewall;
import nz.co.fortytwo.signalk.processor.InputFilterProcessor;
import nz.co.fortytwo.signalk.processor.JsonGetProcessor;
import nz.co.fortytwo.signalk.processor.JsonListProcessor;
import nz.co.fortytwo.signalk.processor.JsonSubscribeProcessor;
import nz.co.fortytwo.signalk.processor.LoggerProcessor;
import nz.co.fortytwo.signalk.processor.MapToJsonProcessor;
import nz.co.fortytwo.signalk.processor.MqttProcessor;
import nz.co.fortytwo.signalk.processor.N2KProcessor;
import nz.co.fortytwo.signalk.processor.NMEA0183ExportProcessor;
import nz.co.fortytwo.signalk.processor.NMEAProcessor;
import nz.co.fortytwo.signalk.processor.OutputFilterProcessor;
import nz.co.fortytwo.signalk.processor.PermissionsProcessor;
import nz.co.fortytwo.signalk.processor.RestApiProcessor;
import nz.co.fortytwo.signalk.processor.RestAuthProcessor;
import nz.co.fortytwo.signalk.processor.RestPathFilterProcessor;
import nz.co.fortytwo.signalk.processor.SaveProcessor;
import nz.co.fortytwo.signalk.processor.SignalkModelProcessor;
import nz.co.fortytwo.signalk.processor.SourceRefToSourceProcessor;
import nz.co.fortytwo.signalk.processor.SourceToSourceRefProcessor;
import nz.co.fortytwo.signalk.processor.StompProcessor;
import nz.co.fortytwo.signalk.processor.TrackProcessor;
import nz.co.fortytwo.signalk.processor.UploadProcessor;
import nz.co.fortytwo.signalk.processor.ValidationProcessor;
import nz.co.fortytwo.signalk.processor.WindProcessor;
import nz.co.fortytwo.signalk.processor.WsSessionProcessor;
import nz.co.fortytwo.signalk.util.ConfigConstants;
import nz.co.fortytwo.signalk.util.SignalKConstants;
/*
 *
 * Copyright (C) 2012-2014 R T Huitema. All Rights Reserved.
 * Web: www.42.co.nz
 * Email: robert@42.co.nz
 * Author: R T Huitema
 *
 * This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING THE
 * WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.
 * 
 * Licensed 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 nz.co.fortytwo.signalk.server;






 
public class SignalkRouteFactory {

	private static Logger logger = LogManager.getLogger(SignalkRouteFactory.class);
	private static Set<String> nameSet = new HashSet<String>();
	/**
	 * Configures a route for all input traffic, which will parse the traffic and update the signalk model
	 * @param routeBuilder
	 * @param input
	 * @param inputFilterProcessor
	 * @param nmeaProcessor
	 * @param aisProcessor
	 * @param signalkModelProcessor
	 * @throws IOException 
	 * @throws Exception 
	 */
	public static void configureInputRoute(RouteBuilder routeBuilder,String input) throws Exception {
		routeBuilder.from(input).id(getName("INPUT"))
			.onException(Exception.class).handled(true).maximumRedeliveries(0)
			.to("log:nz.co.fortytwo.signalk.model.receive?level=ERROR&showException=true&showStackTrace=true")
			.end()
		// dump misc rubbish
		.process(new InputFilterProcessor()).id(getName(InputFilterProcessor.class.getSimpleName()))
		//now filter security
		.process(new IncomingSecurityFirewall()).id(getName(IncomingSecurityFirewall.class.getSimpleName()))
		//swap payloads to storage
		//.process(new StorageProcessor()).id(getName(StorageProcessor.class.getSimpleName()))
		//convert NMEA to signalk
		.process(new NMEAProcessor()).id(getName(NMEAProcessor.class.getSimpleName()))
		//convert AIS to signalk
		.process(new AISProcessor()).id(getName(AISProcessor.class.getSimpleName()))
		//convert n2k
		.process(new N2KProcessor()).id(getName(N2KProcessor.class.getSimpleName()))
		//handle subscribe messages
		.process(new JsonSubscribeProcessor()).id(getName(JsonSubscribeProcessor.class.getSimpleName()))
		//deal with delta format
		.process(new DeltaImportProcessor()).id(getName(DeltaImportProcessor.class.getSimpleName()))
		//deal with full format
		.process(new FullImportProcessor()).id(getName(FullImportProcessor.class.getSimpleName()))
		//make sure we have timestamp/source
		.process(new ValidationProcessor()).id(getName(ValidationProcessor.class.getSimpleName()))
		//record track
		.process(new TrackProcessor()).id(getName(TrackProcessor.class.getSimpleName()))
		//push source to sources and add $source
		.process(new SourceToSourceRefProcessor()).id(getName(SourceToSourceRefProcessor.class.getSimpleName()))
		//strip out according to meta owner,group,others
		.process(new PermissionsProcessor()).id(getName(PermissionsProcessor.class.getSimpleName()))
		//and update signalk model
		.process(new SignalkModelProcessor()).id(getName(SignalkModelProcessor.class.getSimpleName()))
		//we have processed all incoming data now - if there is more left its LIST, GET.
		//handle list
		.process(new JsonListProcessor()).id(getName(JsonListProcessor.class.getSimpleName()))
		//handle get
		.process(new JsonGetProcessor()).id(getName(JsonGetProcessor.class.getSimpleName()));
		//.process(new StorageProcessor()).id(getName(StorageProcessor.class.getSimpleName()));
		
	}
	
	/**
	 * Configures the route for output to websockets
	 * @param routeBuilder
	 * @param input
	 */
	public static void configureWebsocketTxRoute(RouteBuilder routeBuilder ,String input, int port){
		Predicate p1 = routeBuilder.header(ConfigConstants.OUTPUT_TYPE).isEqualTo(ConfigConstants.OUTPUT_WS);
		Predicate p2 = routeBuilder.header(WebsocketConstants.CONNECTION_KEY).isEqualTo(WebsocketConstants.SEND_TO_ALL);
		//from SEDA_WEBSOCKETS
			routeBuilder.from(input).id(getName("Websocket Tx"))
				.onException(Exception.class)
				.handled(true)
				.maximumRedeliveries(0)
				.to("log:nz.co.fortytwo.signalk.model.websocket.tx?level=ERROR&showException=true&showStackTrace=true")
				.end()
			.filter(PredicateBuilder.or(p1, p2))
			.to("skWebsocket://0.0.0.0:"+port+SignalKConstants.SIGNALK_WS).id(getName("Websocket Client"));
		
	}
	/**
	 * Configures the route for input to websockets
	 * @param routeBuilder
	 * @param input
	 * @throws Exception 
	 */
	public static void configureWebsocketRxRoute(RouteBuilder routeBuilder ,String input, int port)  {
		
		WebsocketEndpoint wsEndpoint = (WebsocketEndpoint) routeBuilder.getContext().getEndpoint("skWebsocket://0.0.0.0:"+port+SignalKConstants.SIGNALK_WS);
		wsEndpoint.setEnableJmx(true);
		wsEndpoint.setSessionSupport(true);
		wsEndpoint.setCrossOriginFilterOn(true);
		wsEndpoint.setAllowedOrigins("*");
		
		routeBuilder.from(wsEndpoint).id(getName("Websocket Rx"))
			.onException(Exception.class)
			.handled(true).maximumRedeliveries(0)
			.to("log:nz.co.fortytwo.signalk.model.websocket.rx?level=ERROR&showException=true&showStackTrace=true")
			.end()
		.process(new WsSessionProcessor()).id(getName(WsSessionProcessor.class.getSimpleName()))
		//.to("log:nz.co.fortytwo.signalk.model.websocket.rx?level=INFO&showException=true&showStackTrace=true")
		.to(input).id(getName(RouteManager.SEDA_INPUT));
		
	}
	public static void configureTcpServerRoute(RouteBuilder routeBuilder ,String input, NettyServer nettyServer, String outputType) throws Exception{
		// push out via TCPServer.
		Predicate p1 = routeBuilder.header(ConfigConstants.OUTPUT_TYPE).isEqualTo(outputType);
		Predicate p2 = routeBuilder.header(WebsocketConstants.CONNECTION_KEY).isEqualTo(WebsocketConstants.SEND_TO_ALL);
		routeBuilder.from(input).id(getName("Netty "+outputType+" Server"))
			.onException(Exception.class)
			.handled(true)
			.maximumRedeliveries(0)
			.end()
		.filter(PredicateBuilder.or(p1, p2))
		.process((Processor) nettyServer).id(getName(NettyServer.class.getSimpleName())).end();
			
	}
	
	public static void configureRestRoute(RouteBuilder routeBuilder ,String input, String name)throws IOException{
		routeBuilder.from(input).id(getName(name)) //.setExchangePattern(ExchangePattern.InOut);
			.setExchangePattern(ExchangePattern.InOut)
			.process(new RestApiProcessor())
			.to(ExchangePattern.InOut,RouteManager.SEDA_INPUT)
			.process(new ConfigFilterProcessor(false)).id(getName(ConfigFilterProcessor.class.getSimpleName()))
			.process(new RestPathFilterProcessor()).id(getName(RestPathFilterProcessor.class.getSimpleName()));
		
				
//		routeBuilder.rest(SignalKConstants.SIGNALK_API).id("REST POST Client")
//			    	.post("/")
//					.to("log:nz.co.fortytwo.signalk.client.rest?level=INFO&showException=true&showStackTrace=true")
//					.to("direct:restTest");

		}
	public static void configureRestLoggerRoute(RouteBuilder routeBuilder ,String input, String name)throws IOException{
		routeBuilder.from(input).id(getName(name))
			.setExchangePattern(ExchangePattern.InOut)
			.process(new LoggerProcessor()).id(getName(LoggerProcessor.class.getSimpleName()));
		}
	
	public static void configureRestUploadRoute(RouteBuilder routeBuilder ,String input, String name)throws IOException{
		routeBuilder.rest(input).id(getName(name))
			.post()
			.route().id(getName(name+"Route"))
			.setExchangePattern(ExchangePattern.InOut)
			.process(new UploadProcessor()).id(getName(UploadProcessor.class.getSimpleName()));
		}
	
	public static void configureRestConfigRoute(RouteBuilder routeBuilder ,String input, String name)throws IOException{
		routeBuilder.from(input).id(getName(name)) 
			.setExchangePattern(ExchangePattern.InOut)
			.process(new RestApiProcessor())
			.to(ExchangePattern.InOut,RouteManager.SEDA_INPUT)
			.process(new ConfigFilterProcessor(true)).id(getName(ConfigFilterProcessor.class.getSimpleName()));
		}
	
	public static void configureAuthRoute(RouteBuilder routeBuilder ,String input){
		routeBuilder.from(input).id(getName("REST Authenticate"))
			.setExchangePattern(ExchangePattern.InOut)
			.process(new RestAuthProcessor()).id(getName(RestAuthProcessor.class.getSimpleName()));
		}
	
	public static void configureInstallRoute(RouteBuilder routeBuilder ,String input, String name){
		routeBuilder.from(input).id(getName(name))
			.setExchangePattern(ExchangePattern.InOut)
			.process(new ClientAppProcessor()).id(getName(ClientAppProcessor.class.getSimpleName()));
		}
	
	public static void configureBackgroundTimer(RouteBuilder routeBuilder ,String input){
		routeBuilder.from(input).id(getName("Declination"))
			.process(new AisExpiryProcessor()).id(getName(AisExpiryProcessor.class.getSimpleName()))
			.process(new SaveProcessor()).id(getName(SaveProcessor.class.getSimpleName()))
			.process(new DeclinationProcessor()).id(getName(DeclinationProcessor.class.getSimpleName()))
			.to("log:nz.co.fortytwo.signalk.model.update?level=DEBUG").end();
	}
	
	public static void configureAlarmsTimer(RouteBuilder routeBuilder ,String input){
		routeBuilder.from(input).id(getName("Alarms"))
			.process(new AlarmProcessor()).id(getName(AlarmProcessor.class.getSimpleName()))
			.to("log:nz.co.fortytwo.signalk.model.update?level=DEBUG").end();
	}
	public static void configureNMEA0183Timer(RouteBuilder routeBuilder ,String input){
		routeBuilder.from(input).id(getName("NMEA0183"))
			.process(new NMEA0183ExportProcessor()).id(getName(NMEA0183ExportProcessor.class.getSimpleName()))
			.to("log:nz.co.fortytwo.signalk.model.update?level=DEBUG").end();
	}
	
	public static void configureAnchorWatchTimer(RouteBuilder routeBuilder ,String input){
		routeBuilder.from(input).id(getName("AnchorWatch"))
			.process(new AnchorWatchProcessor()).id(getName(AnchorWatchProcessor.class.getSimpleName()))