#1066 Recommender status info

- started on websocket messaging

inception-recommendation/pom.xml

@@ -251,6 +251,10 @@
       <artifactId>dkpro-core-api-ner-asl</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.wicket</groupId>
+      <artifactId>wicket-native-websocket-core</artifactId>
+    </dependency>
   </dependencies>
   <build>
     <pluginManagement>

inception-recommendation/src/main/java/de/tudarmstadt/ukp/inception/recommendation/service/RecommendationServiceImpl.java

@@ -829,7 +829,7 @@ public Predictions computePredictions(User aUser, Project aProject,
 
                         // Perform the actual prediction
                         recommendationEngine.predict(ctx, predictionCas);
-
+                        
                         // Extract the suggestions from the data which the recommender has written 
                         // into the CAS
                         List<AnnotationSuggestion> suggestions = extractSuggestions(aUser,

inception-recommendation/src/main/java/de/tudarmstadt/ukp/inception/recommendation/sidebar/RecommenderInfoPanel.java

@@ -35,6 +35,9 @@
 import org.apache.wicket.markup.html.panel.Panel;
 import org.apache.wicket.model.IModel;
 import org.apache.wicket.model.LoadableDetachableModel;
+import org.apache.wicket.protocol.ws.api.WebSocketBehavior;
+import org.apache.wicket.protocol.ws.api.WebSocketRequestHandler;
+import org.apache.wicket.protocol.ws.api.message.IWebSocketPushMessage;
 import org.apache.wicket.spring.injection.annot.SpringBean;
 import org.wicketstuff.event.annotation.OnEvent;
 
@@ -111,6 +114,19 @@ protected void populateItem(ListItem<Recommender> item)
         searchResultGroups.setModel(LoadableDetachableModel.of(() -> recommendationService
                 .listEnabledRecommenders(aModel.getObject().getProject())));
         add(searchResultGroups);
+
+        add(new WebSocketBehavior() {
+
+            private static final long serialVersionUID = 1L;
+
+            @Override
+            protected void onPush(WebSocketRequestHandler aHandler, IWebSocketPushMessage aMessage)
+            {
+                // TODO: message specific stuff
+                aHandler.add(searchResultGroups);
+            }
+
+        });
     }
 
     public AnnotatorState getModelObject()

inception-recommendation/src/main/java/de/tudarmstadt/ukp/inception/recommendation/tasks/SelectionTask.java

@@ -48,6 +48,8 @@
 import de.tudarmstadt.ukp.inception.recommendation.event.RecommenderEvaluationResultEvent;
 import de.tudarmstadt.ukp.inception.scheduling.SchedulingService;
 import de.tudarmstadt.ukp.inception.scheduling.Task;
+import de.tudarmstadt.ukp.inception.scheduling.TaskState;
+import de.tudarmstadt.ukp.inception.scheduling.TaskUpdateEvent;
 
 /**
  * This task evaluates all available classification tools for all annotation layers of the current
@@ -178,9 +180,8 @@ protected List<CAS> initialize()
                                 threshold);
                     }
 
-                    appEventPublisher.publishEvent(new RecommenderEvaluationResultEvent(this,
-                            recommender, user.getUsername(), result,
-                            System.currentTimeMillis() - start, activated));
+                    publishFinishedEvents(user, recommender, start, result,
+                            activated);
                 }
                 catch (Throwable e) {
                     log.error("[{}][{}]: Failed", user.getUsername(), recommenderName, e);
@@ -194,6 +195,17 @@ protected List<CAS> initialize()
                 "SelectionTask after activating recommenders"));
     }
 
+    private void publishFinishedEvents(User user, Recommender recommender, long start,
+            EvaluationResult result, boolean activated)
+    {
+        String username = user.getUsername();
+        appEventPublisher.publishEvent(new RecommenderEvaluationResultEvent(this,
+                recommender, username, result,
+                System.currentTimeMillis() - start, activated));
+        getSchedulerCallback().accept(new TaskUpdateEvent(username, TaskState.SELECTION_FINISHED,
+                result.getTrainDataRatio(), recommender.getId(), activated));
+    }
+
     private List<CAS> readCasses(Project aProject, String aUserName)
     {
         List<CAS> casses = new ArrayList<>();

inception-recommendation/src/main/java/de/tudarmstadt/ukp/inception/recommendation/tasks/TrainingTask.java

@@ -53,6 +53,8 @@
 import de.tudarmstadt.ukp.inception.recommendation.api.recommender.RecommenderContext;
 import de.tudarmstadt.ukp.inception.scheduling.SchedulingService;
 import de.tudarmstadt.ukp.inception.scheduling.Task;
+import de.tudarmstadt.ukp.inception.scheduling.TaskState;
+import de.tudarmstadt.ukp.inception.scheduling.TaskUpdateEvent;
 
 /**
  * This consumer trains a new classifier model, if a classification tool was selected before.
@@ -184,6 +186,9 @@ protected List<TrainingDocument> initialize()
 
                     ctx.close();
                     recommendationService.putContext(user, recommender, ctx);
+
+                    getSchedulerCallback().accept(new TaskUpdateEvent(user.getUsername(),
+                            TaskState.TRAINING_FINISHED, recommender.getId()));
                 }
                 catch (Throwable e) {
                     log.info("[{}][{}]: Training failed ({} ms)", user.getUsername(),

inception-scheduling/pom.xml

@@ -80,5 +80,9 @@
       <artifactId>awaitility</artifactId>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>org.apache.wicket</groupId>
+      <artifactId>wicket-native-websocket-core</artifactId>
+    </dependency>
   </dependencies>
 </project>

inception-scheduling/src/main/java/de/tudarmstadt/ukp/inception/scheduling/SchedulingService.java

@@ -22,12 +22,18 @@
 import java.util.List;
 import java.util.concurrent.ThreadPoolExecutor;
 
+import org.apache.wicket.Application;
+import org.apache.wicket.protocol.ws.WebSocketSettings;
+import org.apache.wicket.protocol.ws.api.IWebSocketConnection;
+import org.apache.wicket.protocol.ws.api.registry.IWebSocketConnectionRegistry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
 import org.springframework.context.ApplicationContext;
+import org.springframework.security.core.session.SessionInformation;
+import org.springframework.security.core.session.SessionRegistry;
 import org.springframework.stereotype.Component;
 
 import de.tudarmstadt.ukp.inception.scheduling.config.SchedulingProperties;
@@ -40,6 +46,7 @@ public class SchedulingService
 
     private final ApplicationContext applicationContext;
     private final ThreadPoolExecutor executor;
+    private @Autowired SessionRegistry sessionRegistry;
 
     private final List<Task> runningTasks;
 
@@ -94,6 +101,9 @@ public synchronized void enqueue(Task aTask)
 
         log.debug("Enqueuing task [{}]", aTask);
 
+        // set callback to websocket distribution
+        aTask.setSchedulerCallback(this::distributeWebSocketMessage);
+
         // This autowires the task fields manually.
         AutowireCapableBeanFactory factory = applicationContext.getAutowireCapableBeanFactory();
         factory.autowireBean(aTask);
@@ -122,5 +132,25 @@ public void destroy()
         executor.shutdownNow();
     }
 
-
+    public void distributeWebSocketMessage(TaskUpdateEvent aTaskUpdate)
+    {
+        Application application = Application.get();
+        WebSocketSettings webSocketSettings = WebSocketSettings.Holder.get(application);
+        IWebSocketConnectionRegistry webSocketConnectionRegistry = webSocketSettings
+                .getConnectionRegistry();
+
+        // get all connections for the user
+        List<IWebSocketConnection> userConnections = new ArrayList<>();
+        for (SessionInformation sessionInfo : sessionRegistry
+                .getAllSessions(aTaskUpdate.getUsername(), false)) {
+            userConnections.addAll(webSocketConnectionRegistry.getConnections(application,
+                    sessionInfo.getSessionId()));
+        }
+
+        // send message to all connections
+        for (IWebSocketConnection connection : userConnections) {
+            connection.sendMessage(new TaskWebSocketPushMessage(aTaskUpdate.getProgress(),
+                    aTaskUpdate.getState(), aTaskUpdate.getRecommenderId(), aTaskUpdate.isActive()));
+        }
+    }
 }

inception-scheduling/src/main/java/de/tudarmstadt/ukp/inception/scheduling/Task.java

@@ -20,6 +20,7 @@
 import static org.apache.commons.lang3.Validate.notNull;
 
 import java.util.Objects;
+import java.util.function.Consumer;
 
 import de.tudarmstadt.ukp.clarin.webanno.model.Project;
 import de.tudarmstadt.ukp.clarin.webanno.security.model.User;
@@ -30,6 +31,7 @@ public abstract class Task
     private final User user;
     private final Project project;
     private final String trigger;
+    private Consumer<TaskUpdateEvent> schedulerCallback;
 
     public Task(User aUser, Project aProject, String aTrigger)
     {
@@ -85,4 +87,14 @@ public boolean equals(Object o) {
     public int hashCode() {
         return Objects.hash(user, project);
     }
+
+    protected void setSchedulerCallback(Consumer<TaskUpdateEvent> aCallback)
+    {
+        schedulerCallback = aCallback;
+    }
+
+    protected Consumer<TaskUpdateEvent> getSchedulerCallback()
+    {
+        return schedulerCallback;
+    }
 }

inception-scheduling/src/main/java/de/tudarmstadt/ukp/inception/scheduling/TaskState.java

@@ -0,0 +1,6 @@
+package de.tudarmstadt.ukp.inception.scheduling;
+
+public enum TaskState
+{
+    TRAINING_STARTED, TRAINING_FINISHED, SELECTION_STARTED, SELECTION_FINISHED
+}

inception-scheduling/src/main/java/de/tudarmstadt/ukp/inception/scheduling/TaskUpdateEvent.java

@@ -0,0 +1,53 @@
+package de.tudarmstadt.ukp.inception.scheduling;
+
+
+public class TaskUpdateEvent
+{
+    private final String username;
+    private final double progress;
+    private final TaskState state;
+    private final long recommenderId;
+    private final boolean active;
+
+    //TODO: add evaluationResult?
+
+    public TaskUpdateEvent(String aName, TaskState aState, double aProgress, long aRecommenderId,
+            boolean aActive)
+    {
+        username = aName;
+        state = aState;
+        progress = aProgress;
+        recommenderId = aRecommenderId;
+        active = aActive;
+    }
+
+    public TaskUpdateEvent(String aUsername, TaskState aState, long aRecommenderId)
+    {
+        this(aUsername, aState, 1, aRecommenderId, true);
+    }
+
+    public String getUsername()
+    {
+        return username;
+    }
+
+    public double getProgress()
+    {
+        return progress;
+    }
+
+    public TaskState getState()
+    {
+        return state;
+    }
+
+    public long getRecommenderId()
+    {
+        return recommenderId;
+    }
+
+    public boolean isActive()
+    {
+        return active;
+    }
+}

inception-scheduling/src/main/java/de/tudarmstadt/ukp/inception/scheduling/TaskWebSocketPushMessage.java

@@ -0,0 +1,16 @@
+package de.tudarmstadt.ukp.inception.scheduling;
+
+import org.apache.wicket.protocol.ws.api.message.IWebSocketPushMessage;
+
+public class TaskWebSocketPushMessage
+    implements IWebSocketPushMessage
+{
+    private static final long serialVersionUID = 1L;
+
+    public TaskWebSocketPushMessage(double aProgress, TaskState aState, long aRecommenderId,
+            boolean aActive)
+    {
+        // TODO Auto-generated constructor stub
+    }
+
+}