solr/solrj/src/java/org/apache/solr/client/solrj/io/stream/DaemonStream.java - lucene-solr - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.solr.client.solrj.io.stream;

 import java.io.IOException;
 import java.lang.Thread.State;
 import java.lang.invoke.MethodHandles;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.atomic.AtomicLong;

 import org.apache.solr.client.solrj.io.Tuple;
 import org.apache.solr.client.solrj.io.comp.StreamComparator;
 import org.apache.solr.client.solrj.io.stream.expr.Explanation;
 import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
 import org.apache.solr.common.util.ExecutorUtil;
 import org.apache.solr.common.util.SolrNamedThreadFactory;
 import org.apache.solr.client.solrj.io.stream.expr.Expressible;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
 import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
 import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import static org.apache.solr.common.params.CommonParams.ID;

 /**
  * @since 6.0.0
  */
 public class DaemonStream extends TupleStream implements Expressible {

   private TupleStream tupleStream;
   private StreamRunner streamRunner;
   private ArrayBlockingQueue<Tuple> queue;
   private int queueSize;
   private boolean eatTuples;
   private AtomicLong iterations = new AtomicLong();
   private long startTime;
   private long stopTime;
   private Exception exception;
   private long runInterval;
   private String id;
   private Map<String, DaemonStream> daemons;
   private boolean terminate;
   private boolean closed = false;
   private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

   public DaemonStream(StreamExpression expression, StreamFactory factory) throws IOException{

     List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);

     TupleStream tupleStream = factory.constructStream(streamExpressions.get(0));

     StreamExpressionNamedParameter idExpression = factory.getNamedOperand(expression, ID);
     StreamExpressionNamedParameter runExpression = factory.getNamedOperand(expression, "runInterval");
     StreamExpressionNamedParameter queueExpression = factory.getNamedOperand(expression, "queueSize");
     StreamExpressionNamedParameter terminateExpression = factory.getNamedOperand(expression, "terminate");


     String id = null;
     long runInterval = 0L;
     int queueSize = 0;
     boolean terminate = false;

     if(idExpression == null) {
       throw new IOException("Invalid expression id parameter expected");
     } else {
       id = ((StreamExpressionValue) idExpression.getParameter()).getValue();
     }

     if(runExpression == null) {
       runInterval = 2000;
     } else {
       runInterval = Long.parseLong(((StreamExpressionValue) runExpression.getParameter()).getValue());
     }

     if(queueExpression != null) {
        queueSize= Integer.parseInt(((StreamExpressionValue) queueExpression.getParameter()).getValue());
     }

     if(terminateExpression != null) {
       terminate = Boolean.parseBoolean(((StreamExpressionValue) terminateExpression.getParameter()).getValue());
     }

     if(1 != streamExpressions.size()){
       throw new IOException(String.format(Locale.ROOT,"Invalid expression %s - expecting a single stream but found %d",expression, streamExpressions.size()));
     }

     init(tupleStream, id, runInterval, queueSize, terminate);
   }

   public DaemonStream(TupleStream tupleStream, String id, long runInterval, int queueSize, boolean terminate) {
     init(tupleStream, id, runInterval, queueSize, terminate);
   }

   public DaemonStream(TupleStream tupleStream, String id, long runInterval, int queueSize) {
     this(tupleStream, id, runInterval, queueSize, false);
   }

   @Override
   public StreamExpression toExpression(StreamFactory factory) throws IOException{
     return toExpression(factory, true);
   }

   private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
     // function name
     StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));

     if(includeStreams){
       // streams
       if(tupleStream instanceof Expressible){
         expression.addParameter(((Expressible)tupleStream).toExpression(factory));
       } else {
         throw new IOException("This UniqueStream contains a non-expressible TupleStream - it cannot be converted to an expression");
       }
     }
     else{
       expression.addParameter("<stream>");
     }

     expression.addParameter(new StreamExpressionNamedParameter(ID, id));
     expression.addParameter(new StreamExpressionNamedParameter("runInterval", Long.toString(runInterval)));
     expression.addParameter(new StreamExpressionNamedParameter("queueSize", Integer.toString(queueSize)));
     expression.addParameter(new StreamExpressionNamedParameter("terminate", Boolean.toString(terminate)));

     return expression;
   }

   @Override
   public Explanation toExplanation(StreamFactory factory) throws IOException {

     return new StreamExplanation(getStreamNodeId().toString())
       .withChildren(new Explanation[] {
         tupleStream.toExplanation(factory)
       })
       .withFunctionName(factory.getFunctionName(this.getClass()))
       .withImplementingClass(this.getClass().getName())
       .withExpressionType(ExpressionType.STREAM_DECORATOR)
       .withExpression(toExpression(factory, false).toString());
   }

   public int remainingCapacity() {
     return this.queue.remainingCapacity();
   }

   public void init(TupleStream tupleStream, String id, long runInterval, int queueSize) {
     init(tupleStream, id, runInterval, queueSize, false);
   }

   @SuppressWarnings({"unchecked", "rawtypes"})
   public void init(TupleStream tupleStream, String id, long runInterval, int queueSize, boolean terminate) {
     this.tupleStream = tupleStream;
     this.id = id;
     this.runInterval = runInterval;
     this.queueSize = queueSize;
     this.terminate = terminate;

     if(queueSize > 0) {
       queue = new ArrayBlockingQueue(queueSize);
       eatTuples = false;
     } else {
       eatTuples = true;
     }
   }

   public int hashCode() {
     return id.hashCode();
   }

   public boolean equals(Object o) {
     if(o instanceof DaemonStream) {
       return id.equals(((DaemonStream)o).id);
     }
     return false;
   }

   public String getId() {
     return id;
   }

   public void open() throws IOException {
     if (this.streamRunner != null && this.closed == false) {
       log.error("There is already a running daemon named '{}', no action taken", id);
       throw new IOException("There is already an open daemon named '" + id + "', no action taken.");
     }
     this.closed = false;
     this.streamRunner = new StreamRunner(runInterval, id);
     ExecutorService service = ExecutorUtil.newMDCAwareSingleThreadExecutor(new SolrNamedThreadFactory("DaemonStream-" + id));
     try {
       service.submit(this.streamRunner);
     }
     finally {
       service.shutdown();
     }
   }

   public Tuple read() throws IOException {
     try {
       return queue.take();
     } catch (Exception e) {
       throw new IOException(e);
     }
   }

   public StreamComparator getStreamSort() {
     return tupleStream.getStreamSort();
   }

   public void setStreamContext(StreamContext streamContext) {
     this.tupleStream.setStreamContext(streamContext);
   }

   public void shutdown() {
     streamRunner.setShutdown(true);
   }

   public void close() {
     if(closed) {
       return;
     }
     if (streamRunner != null) {
       streamRunner.setShutdown(true);
     }
     this.closed = true;
   }

   public List<TupleStream> children() {
     List<TupleStream> children = new ArrayList<>();
     children.add(tupleStream);
     return children;
   }

   public synchronized Tuple getInfo() {
     Tuple tuple = new Tuple();
     tuple.put(ID, id);
     tuple.put("startTime", startTime);
     tuple.put("stopTime", stopTime);
     tuple.put("iterations", iterations.get());
     tuple.put("state", streamRunner.getState().toString());
     if (exception != null) {
       tuple.put("exception", exception.getMessage());
     }

     return tuple;
   }

   public void setDaemons(Map<String, DaemonStream> daemons) {
     this.daemons = daemons;
   }

   private synchronized void setStartTime(long startTime) {
     this.startTime = startTime;
   }

   private synchronized void setStopTime(long stopTime) {
     this.stopTime = stopTime;
   }

   private class StreamRunner implements Runnable {

     private long sleepMillis = 1000;
     private long runInterval;
     private long lastRun;
     private String id;

     // a reference to the Thread that is executing the stream to track its state
     private volatile Thread executingThread;
     private boolean shutdown;

     public StreamRunner(long runInterval, String id) {
       this.runInterval = runInterval;
       this.id = id;
     }

     public synchronized void setShutdown(boolean shutdown) {
       this.shutdown = shutdown;
     }

     public synchronized boolean getShutdown() {
       return shutdown;
     }

     public State getState() {
       if (executingThread == null) {
         if (shutdown) {
           return Thread.State.TERMINATED;
         } else {
           return Thread.State.NEW;
         }
       } else {
         return executingThread.getState();
       }
     }

     public void run() {
       executingThread = Thread.currentThread();
       try {
         stream();
       } finally {
         setShutdown(true);
         executingThread = null;
       }
     }

     private void stream() {
       int errors = 0;
       setStartTime(new Date().getTime());
       OUTER:
       while (!getShutdown()) {
         long now = new Date().getTime();
         if ((now - lastRun) > this.runInterval) {
           lastRun = now;
           try {
             tupleStream.open();
             INNER:
             while (true) {
               Tuple tuple = tupleStream.read();
               if (tuple.EOF) {
                 errors = 0; // Reset errors on successful run.
                 if (tuple.getFields().containsKey("sleepMillis")) {
                   this.sleepMillis = tuple.getLong("sleepMillis");

                   if(terminate && sleepMillis > 0) {
                     //TopicStream provides sleepMillis > 0 if the last run had no Tuples.
                     //This means the topic queue is empty. Time to terminate.
                     //Remove ourselves from the daemons map.
                     if(daemons != null) {
                       daemons.remove(id);
                     }
                     //Break out of the thread loop and end the run.
                     break OUTER;
                   }

                   this.runInterval = -1;
                 }
                 break INNER;
               } else if (!eatTuples) {
                 try {
                   queue.put(tuple);
                 } catch (InterruptedException e) {
                   break OUTER;
                 }
               }
             }
           } catch (IOException e) {
             exception = e;
             log.error("Error in DaemonStream: {}", id, e);
             ++errors;
             if (errors > 100) {
               log.error("Too many consecutive errors. Stopping DaemonStream: {}", id);
               break OUTER;
             }
           } catch (Throwable t) {
             log.error("Fatal Error in DaemonStream: {}",  id, t);
             //For anything other then IOException break out of the loop and shutdown the thread.
             break OUTER;
           } finally {
             try {
               tupleStream.close();
             } catch (IOException e1) {
               if (exception == null) {
                 exception = e1;
                 log.error("Error in DaemonStream: {}", id, e1);
                 break OUTER;
               }
             }
           }
         }
         iterations.incrementAndGet();

         if (sleepMillis > 0) {
           try {
             Thread.sleep(sleepMillis);
           } catch (InterruptedException e) {
             log.error("Error in DaemonStream:{}", id, e);
             break OUTER;
           }
         }
       }

       if(!eatTuples) {
         try {
           queue.put(Tuple.EOF());
         } catch (InterruptedException e) {
           log.error("Error in DaemonStream:{}", id, e);
         }
       }
       setStopTime(new Date().getTime());
     }
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.solr.client.solrj.io.stream;

	import java.io.IOException;
	import java.lang.Thread.State;
	import java.lang.invoke.MethodHandles;
	import java.util.ArrayList;
	import java.util.Date;
	import java.util.List;
	import java.util.Locale;
	import java.util.Map;
	import java.util.concurrent.ArrayBlockingQueue;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.atomic.AtomicLong;

	import org.apache.solr.client.solrj.io.Tuple;
	import org.apache.solr.client.solrj.io.comp.StreamComparator;
	import org.apache.solr.client.solrj.io.stream.expr.Explanation;
	import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
	import org.apache.solr.common.util.ExecutorUtil;
	import org.apache.solr.common.util.SolrNamedThreadFactory;
	import org.apache.solr.client.solrj.io.stream.expr.Expressible;
	import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
	import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
	import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
	import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
	import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import static org.apache.solr.common.params.CommonParams.ID;

	/**
	* @since 6.0.0
	*/
	public class DaemonStream extends TupleStream implements Expressible {

	private TupleStream tupleStream;
	private StreamRunner streamRunner;
	private ArrayBlockingQueue<Tuple> queue;
	private int queueSize;
	private boolean eatTuples;
	private AtomicLong iterations = new AtomicLong();
	private long startTime;
	private long stopTime;
	private Exception exception;
	private long runInterval;
	private String id;
	private Map<String, DaemonStream> daemons;
	private boolean terminate;
	private boolean closed = false;
	private static final Logger log = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

	public DaemonStream(StreamExpression expression, StreamFactory factory) throws IOException{

	List<StreamExpression> streamExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, TupleStream.class);

	TupleStream tupleStream = factory.constructStream(streamExpressions.get(0));

	StreamExpressionNamedParameter idExpression = factory.getNamedOperand(expression, ID);
	StreamExpressionNamedParameter runExpression = factory.getNamedOperand(expression, "runInterval");
	StreamExpressionNamedParameter queueExpression = factory.getNamedOperand(expression, "queueSize");
	StreamExpressionNamedParameter terminateExpression = factory.getNamedOperand(expression, "terminate");


	String id = null;
	long runInterval = 0L;
	int queueSize = 0;
	boolean terminate = false;

	if(idExpression == null) {
	throw new IOException("Invalid expression id parameter expected");
	} else {
	id = ((StreamExpressionValue) idExpression.getParameter()).getValue();
	}

	if(runExpression == null) {
	runInterval = 2000;
	} else {
	runInterval = Long.parseLong(((StreamExpressionValue) runExpression.getParameter()).getValue());
	}

	if(queueExpression != null) {
	queueSize= Integer.parseInt(((StreamExpressionValue) queueExpression.getParameter()).getValue());
	}

	if(terminateExpression != null) {
	terminate = Boolean.parseBoolean(((StreamExpressionValue) terminateExpression.getParameter()).getValue());
	}

	if(1 != streamExpressions.size()){
	throw new IOException(String.format(Locale.ROOT,"Invalid expression %s - expecting a single stream but found %d",expression, streamExpressions.size()));
	}

	init(tupleStream, id, runInterval, queueSize, terminate);
	}

	public DaemonStream(TupleStream tupleStream, String id, long runInterval, int queueSize, boolean terminate) {
	init(tupleStream, id, runInterval, queueSize, terminate);
	}

	public DaemonStream(TupleStream tupleStream, String id, long runInterval, int queueSize) {
	this(tupleStream, id, runInterval, queueSize, false);
	}

	@Override
	public StreamExpression toExpression(StreamFactory factory) throws IOException{
	return toExpression(factory, true);
	}

	private StreamExpression toExpression(StreamFactory factory, boolean includeStreams) throws IOException {
	// function name
	StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));

	if(includeStreams){
	// streams
	if(tupleStream instanceof Expressible){
	expression.addParameter(((Expressible)tupleStream).toExpression(factory));
	} else {
	throw new IOException("This UniqueStream contains a non-expressible TupleStream - it cannot be converted to an expression");
	}
	}
	else{
	expression.addParameter("<stream>");
	}

	expression.addParameter(new StreamExpressionNamedParameter(ID, id));
	expression.addParameter(new StreamExpressionNamedParameter("runInterval", Long.toString(runInterval)));
	expression.addParameter(new StreamExpressionNamedParameter("queueSize", Integer.toString(queueSize)));
	expression.addParameter(new StreamExpressionNamedParameter("terminate", Boolean.toString(terminate)));

	return expression;
	}

	@Override
	public Explanation toExplanation(StreamFactory factory) throws IOException {

	return new StreamExplanation(getStreamNodeId().toString())
	.withChildren(new Explanation[] {
	tupleStream.toExplanation(factory)
	})
	.withFunctionName(factory.getFunctionName(this.getClass()))
	.withImplementingClass(this.getClass().getName())
	.withExpressionType(ExpressionType.STREAM_DECORATOR)
	.withExpression(toExpression(factory, false).toString());
	}

	public int remainingCapacity() {
	return this.queue.remainingCapacity();
	}

	public void init(TupleStream tupleStream, String id, long runInterval, int queueSize) {
	init(tupleStream, id, runInterval, queueSize, false);
	}

	@SuppressWarnings({"unchecked", "rawtypes"})
	public void init(TupleStream tupleStream, String id, long runInterval, int queueSize, boolean terminate) {
	this.tupleStream = tupleStream;
	this.id = id;
	this.runInterval = runInterval;
	this.queueSize = queueSize;
	this.terminate = terminate;

	if(queueSize > 0) {
	queue = new ArrayBlockingQueue(queueSize);
	eatTuples = false;
	} else {
	eatTuples = true;
	}
	}

	public int hashCode() {
	return id.hashCode();
	}

	public boolean equals(Object o) {
	if(o instanceof DaemonStream) {
	return id.equals(((DaemonStream)o).id);
	}
	return false;
	}

	public String getId() {
	return id;
	}

	public void open() throws IOException {
	if (this.streamRunner != null && this.closed == false) {
	log.error("There is already a running daemon named '{}', no action taken", id);
	throw new IOException("There is already an open daemon named '" + id + "', no action taken.");
	}
	this.closed = false;
	this.streamRunner = new StreamRunner(runInterval, id);
	ExecutorService service = ExecutorUtil.newMDCAwareSingleThreadExecutor(new SolrNamedThreadFactory("DaemonStream-" + id));
	try {
	service.submit(this.streamRunner);
	}
	finally {
	service.shutdown();
	}
	}

	public Tuple read() throws IOException {
	try {
	return queue.take();
	} catch (Exception e) {
	throw new IOException(e);
	}
	}

	public StreamComparator getStreamSort() {
	return tupleStream.getStreamSort();
	}

	public void setStreamContext(StreamContext streamContext) {
	this.tupleStream.setStreamContext(streamContext);
	}

	public void shutdown() {
	streamRunner.setShutdown(true);
	}

	public void close() {
	if(closed) {
	return;
	}
	if (streamRunner != null) {
	streamRunner.setShutdown(true);
	}
	this.closed = true;
	}

	public List<TupleStream> children() {
	List<TupleStream> children = new ArrayList<>();
	children.add(tupleStream);
	return children;
	}

	public synchronized Tuple getInfo() {
	Tuple tuple = new Tuple();
	tuple.put(ID, id);
	tuple.put("startTime", startTime);
	tuple.put("stopTime", stopTime);
	tuple.put("iterations", iterations.get());
	tuple.put("state", streamRunner.getState().toString());
	if (exception != null) {
	tuple.put("exception", exception.getMessage());
	}

	return tuple;
	}

	public void setDaemons(Map<String, DaemonStream> daemons) {
	this.daemons = daemons;
	}

	private synchronized void setStartTime(long startTime) {
	this.startTime = startTime;
	}

	private synchronized void setStopTime(long stopTime) {
	this.stopTime = stopTime;
	}

	private class StreamRunner implements Runnable {

	private long sleepMillis = 1000;
	private long runInterval;
	private long lastRun;
	private String id;

	// a reference to the Thread that is executing the stream to track its state
	private volatile Thread executingThread;
	private boolean shutdown;

	public StreamRunner(long runInterval, String id) {
	this.runInterval = runInterval;
	this.id = id;
	}

	public synchronized void setShutdown(boolean shutdown) {
	this.shutdown = shutdown;
	}

	public synchronized boolean getShutdown() {
	return shutdown;
	}

	public State getState() {
	if (executingThread == null) {
	if (shutdown) {
	return Thread.State.TERMINATED;
	} else {
	return Thread.State.NEW;
	}
	} else {
	return executingThread.getState();
	}
	}

	public void run() {
	executingThread = Thread.currentThread();
	try {
	stream();
	} finally {
	setShutdown(true);
	executingThread = null;
	}
	}

	private void stream() {
	int errors = 0;
	setStartTime(new Date().getTime());
	OUTER:
	while (!getShutdown()) {
	long now = new Date().getTime();
	if ((now - lastRun) > this.runInterval) {
	lastRun = now;
	try {
	tupleStream.open();
	INNER:
	while (true) {
	Tuple tuple = tupleStream.read();
	if (tuple.EOF) {
	errors = 0; // Reset errors on successful run.
	if (tuple.getFields().containsKey("sleepMillis")) {
	this.sleepMillis = tuple.getLong("sleepMillis");

	if(terminate && sleepMillis > 0) {
	//TopicStream provides sleepMillis > 0 if the last run had no Tuples.
	//This means the topic queue is empty. Time to terminate.
	//Remove ourselves from the daemons map.
	if(daemons != null) {
	daemons.remove(id);
	}
	//Break out of the thread loop and end the run.
	break OUTER;
	}

	this.runInterval = -1;
	}
	break INNER;
	} else if (!eatTuples) {
	try {
	queue.put(tuple);
	} catch (InterruptedException e) {
	break OUTER;
	}
	}
	}
	} catch (IOException e) {
	exception = e;
	log.error("Error in DaemonStream: {}", id, e);
	++errors;
	if (errors > 100) {
	log.error("Too many consecutive errors. Stopping DaemonStream: {}", id);
	break OUTER;
	}
	} catch (Throwable t) {
	log.error("Fatal Error in DaemonStream: {}", id, t);
	//For anything other then IOException break out of the loop and shutdown the thread.
	break OUTER;
	} finally {
	try {
	tupleStream.close();
	} catch (IOException e1) {
	if (exception == null) {
	exception = e1;
	log.error("Error in DaemonStream: {}", id, e1);
	break OUTER;
	}
	}
	}
	}
	iterations.incrementAndGet();

	if (sleepMillis > 0) {
	try {
	Thread.sleep(sleepMillis);
	} catch (InterruptedException e) {
	log.error("Error in DaemonStream:{}", id, e);
	break OUTER;
	}
	}
	}

	if(!eatTuples) {
	try {
	queue.put(Tuple.EOF());
	} catch (InterruptedException e) {
	log.error("Error in DaemonStream:{}", id, e);
	}
	}
	setStopTime(new Date().getTime());
	}
	}
	}