/**
 * Copyright 2013 Microsoft.
 *
 * 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 com.microsoft.wake.impl;

import javax.inject.Inject;


import com.microsoft.tang.annotations.Parameter;
import com.microsoft.wake.AbstractEStage;
import com.microsoft.wake.EventHandler;
import com.microsoft.wake.StageConfiguration.StageHandler;
import com.microsoft.wake.StageConfiguration.StageName;

/**
 * Stage that synchronously executes an event handler
 * 
 * @param <T> type
 */
public final class SyncStage<T> extends AbstractEStage<T> {

  private final EventHandler<T> handler;

  /**
   * Constructs a synchronous stage
   * 
   * @param handler the event handler
   */
  @Inject
  public SyncStage(final @Parameter(StageHandler.class) EventHandler<T> handler) {
    this(handler.getClass().getName(), handler);
  }
  
  /**
   * Constructs a synchronous stage
   * 
   * @name name the stage name
   * @param handler the event handler
   */
  @Inject
  public SyncStage(final @Parameter(StageName.class) String name, 
      final @Parameter(StageHandler.class) EventHandler<T> handler) {
    super(name);
    this.handler = handler;
    StageManager.instance().register(this);
  }

	
  /**
   * Invokes the handler for the event
   * 
   * @param value an event
   */
  @Override
  public void onNext(T value) {
    beforeOnNext();
	handler.onNext(value);
	afterOnNext();
  }

  /**
   * Closes resources
   * 
   * @throws Exception
   */
  @Override
  public void close() throws Exception {
  }

}

@@ -623,16 +623,82 @@
 .impl;%0A%0A
+import java.util.logging.Level;%0Aimport java.util.logging.Logger;%0A%0A
 import j
@@ -718,17 +718,16 @@
 nject;%0A%0A
-%0A
 import c
@@ -841,32 +841,91 @@
 e.EventHandler;%0A
+import com.microsoft.wake.StageConfiguration.ErrorHandler;%0A
 import com.micro
@@ -1184,31 +1184,167 @@
 ate 
-final EventHandler%3CT%3E h
+static final Logger LOG = Logger.getLogger(SyncStage.class.getName());%0A%0A  private final EventHandler%3CT%3E handler;%0A  private final EventHandler%3CThrowable%3E errorH
 andl
@@ -1579,16 +1579,22 @@
  handler
+, null
 );%0A  %7D%0A 
@@ -1861,24 +1861,454 @@
  handler) %7B%0A
+    this(name, handler, null);%0A  %7D%0A%0A  /**%0A   * Constructs a synchronous stage%0A   * %0A   * @name name the stage name%0A   * @param handler the event handler%0A   * @param errorHandler the error handler%0A   */%0A  @Inject%0A  public SyncStage(final @Parameter(StageName.class) String name, %0A      final @Parameter(StageHandler.class) EventHandler%3CT%3E handler,%0A      final @Parameter(ErrorHandler.class) EventHandler%3CThrowable%3E errorHandler) %7B%0A
     super(na
@@ -2340,16 +2340,54 @@
 andler;%0A
+    this.errorHandler = errorHandler;%0A
     Stag
@@ -2426,17 +2426,16 @@
 s);%0A  %7D%0A
-%0A
 %09%0A  /**%0A
@@ -2497,18 +2497,19 @@
 m value 
-an
+the
  event%0A 
@@ -2546,16 +2546,22 @@
  onNext(
+final 
 T value)
@@ -2583,16 +2583,30 @@
 Next();%0A
+    try %7B%0A    
 %09handler
@@ -2621,17 +2621,227 @@
 value);%0A
-%09
+    %7D catch (Throwable t) %7B%0A    %09if (errorHandler != null) %7B%0A    %09%09errorHandler.onNext(t);%0A    %09%7D else %7B%0A    %09  LOG.log(Level.SEVERE, name + %22 Exception from event handler%22, t);%0A    %09  throw t;%0A    %09%7D%0A    %7D%0A    
 afterOnN

Add ErrorHandler to SyncStage