Merge branch steve-community/steve 'master' into feature/ocpp16-security

# Conflicts:
#	src/main/java/de/rwth/idsg/steve/service/CentralSystemService16_Service.java
#	src/test/java/de/rwth/idsg/steve/utils/__DatabasePreparer__.java

Author: goekay

pom.xml

@@ -9,7 +9,7 @@
     <parent>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-parent</artifactId>
-        <version>3.5.6</version>
+        <version>3.5.7</version>
         <relativePath/> <!-- lookup parent from repository -->
     </parent>
 
@@ -170,7 +170,7 @@
             <plugin>
                 <groupId>com.github.spotbugs</groupId>
                 <artifactId>spotbugs-maven-plugin</artifactId>
-                <version>4.9.4.2</version>
+                <version>4.9.8.1</version>
                 <configuration>
                     <failOnError>false</failOnError>
                     <onlyAnalyze>de.rwth.idsg.steve.-</onlyAnalyze>

src/main/java/de/rwth/idsg/steve/NotificationFeature.java

@@ -21,6 +21,13 @@
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static de.rwth.idsg.steve.utils.StringUtils.joinByComma;
+import static de.rwth.idsg.steve.utils.StringUtils.splitByComma;
+
 /**
  * @author Sevket Goekay <[email protected]>
  * @since 22.01.2016
@@ -30,17 +37,31 @@ public enum NotificationFeature {
 
     // Ocpp related
     //
-    OcppStationBooted(" a charging station sends a boot notification (Note: This activates notifications about failed connection attempts for unregistered JSON stations, as well)"),
-    OcppStationStatusFailure(" a connector gets faulted"),
-    OcppStationWebSocketConnected(" a JSON charging station connects"),
-    OcppStationWebSocketDisconnected(" a JSON charging station disconnects"),
-    OcppTransactionStarted(" a charging station starts a transaction"),
-    OcppTransactionEnded(" a charging station ends a transaction");
+    OcppStationBooted(false, " a charging station sends a boot notification (Note: This activates notifications about failed connection attempts for unregistered JSON stations, as well)"),
+    OcppStationStatusFailure(true, " a connector gets faulted"),
+    OcppStationWebSocketConnected(false, " a JSON charging station connects"),
+    OcppStationWebSocketDisconnected(false, " a JSON charging station disconnects"),
+    OcppTransactionStarted(true, " a charging station starts a transaction"),
+    OcppStationStatusSuspendedEV(true, " a EV suspended charging"),
+    OcppTransactionEnded(true, " a charging station ends a transaction");
 
+    private static final List<NotificationFeature> userNotificationFeatures = Arrays.stream(NotificationFeature.values())
+        .filter(NotificationFeature::isForUser)
+        .collect(Collectors.toList());
+
+    /**
+     * Whether this notification type is intended for end-users as well.
+     */
+    @Getter
+    private final boolean forUser;
 
     @Getter
     private final String text;
 
+    public static List<NotificationFeature> getUserValues() {
+        return userNotificationFeatures;
+    }
+
     public static NotificationFeature fromName(String v) {
         for (NotificationFeature c: NotificationFeature.values()) {
             if (c.name().equalsIgnoreCase(v)) {
@@ -49,4 +70,14 @@ public static NotificationFeature fromName(String v) {
         }
         throw new IllegalArgumentException(v);
     }
+
+    public static List<NotificationFeature> splitFeatures(String str) {
+        return splitByComma(str).stream()
+            .map(NotificationFeature::fromName)
+            .collect(Collectors.toList());
+    }
+
+    public static String joinFeatures(List<NotificationFeature> enablesFeatures) {
+        return joinByComma(enablesFeatures);
+    }
 }

src/main/java/de/rwth/idsg/steve/config/BeanConfiguration.java

@@ -43,6 +43,7 @@
 import org.springframework.format.support.FormattingConversionService;
 import org.springframework.http.converter.HttpMessageConverter;
 import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
+import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -67,6 +68,7 @@
 @Configuration
 @EnableWebMvc
 @EnableScheduling
+@EnableAsync
 @ComponentScan("de.rwth.idsg.steve")
 public class BeanConfiguration implements WebMvcConfigurer {
 
@@ -130,28 +132,29 @@ public DSLContext dslContext(DataSource dataSource,
         return DSL.using(conf);
     }
 
-    @Bean(destroyMethod = "close")
-    public DelegatingTaskScheduler asyncTaskScheduler() {
+    @Bean
+    public ThreadPoolTaskScheduler taskScheduler() {
         ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
         scheduler.setPoolSize(5);
         scheduler.setThreadNamePrefix("SteVe-TaskScheduler-");
         scheduler.setWaitForTasksToCompleteOnShutdown(true);
         scheduler.setAwaitTerminationSeconds(30);
         scheduler.initialize();
 
-        return new DelegatingTaskScheduler(scheduler);
+        return scheduler;
     }
 
-    @Bean(destroyMethod = "close")
-    public DelegatingTaskExecutor asyncTaskExecutor() {
+    @Bean
+    @Primary
+    public ThreadPoolTaskExecutor taskExecutor() {
         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
         executor.setCorePoolSize(5);
         executor.setThreadNamePrefix("SteVe-TaskExecutor-");
         executor.setWaitForTasksToCompleteOnShutdown(true);
         executor.setAwaitTerminationSeconds(30);
         executor.initialize();
 
-        return new DelegatingTaskExecutor(executor);
+        return executor;
     }
 
     /**

src/main/java/de/rwth/idsg/steve/config/DelegatingTaskScheduler.java

@@ -23,7 +23,7 @@
 import de.rwth.idsg.steve.ocpp.ws.ocpp12.Ocpp12WebSocketEndpoint;
 import de.rwth.idsg.steve.ocpp.ws.ocpp15.Ocpp15WebSocketEndpoint;
 import de.rwth.idsg.steve.ocpp.ws.ocpp16.Ocpp16WebSocketEndpoint;
-import de.rwth.idsg.steve.service.ChargePointRegistrationService;
+import de.rwth.idsg.steve.service.ChargePointService;
 import de.rwth.idsg.steve.web.validation.ChargeBoxIdValidator;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -48,7 +48,7 @@
 @Slf4j
 public class WebSocketConfiguration implements WebSocketConfigurer {
 
-    private final ChargePointRegistrationService chargePointRegistrationService;
+    private final ChargePointService chargePointService;
     private final ChargeBoxIdValidator chargeBoxIdValidator;
     private final Ocpp12WebSocketEndpoint ocpp12WebSocketEndpoint;
     private final Ocpp15WebSocketEndpoint ocpp15WebSocketEndpoint;
@@ -65,7 +65,7 @@ public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
             chargeBoxIdValidator,
             handshakeHandler(),
             Lists.newArrayList(ocpp16WebSocketEndpoint, ocpp15WebSocketEndpoint, ocpp12WebSocketEndpoint),
-            chargePointRegistrationService
+            chargePointService
         );
 
         registry.addHandler(handshakeHandler.getDummyWebSocketHandler(), PATH_INFIX + "*")

src/main/java/de/rwth/idsg/steve/config/WebSocketConfiguration.java

@@ -18,11 +18,10 @@
  */
 package de.rwth.idsg.steve.ocpp.soap;
 
-import de.rwth.idsg.steve.config.DelegatingTaskExecutor;
 import de.rwth.idsg.steve.ocpp.OcppProtocol;
 import de.rwth.idsg.steve.repository.OcppServerRepository;
 import de.rwth.idsg.steve.repository.impl.ChargePointRepositoryImpl;
-import de.rwth.idsg.steve.service.ChargePointRegistrationService;
+import de.rwth.idsg.steve.service.ChargePointService;
 import lombok.extern.slf4j.Slf4j;
 import ocpp.cs._2015._10.RegistrationStatus;
 import org.apache.cxf.binding.soap.Soap12;
@@ -37,6 +36,7 @@
 import org.apache.cxf.ws.addressing.AddressingProperties;
 import org.apache.cxf.ws.addressing.ContextUtils;
 import org.apache.cxf.ws.addressing.EndpointReferenceType;
+import org.springframework.core.task.TaskExecutor;
 import org.springframework.stereotype.Component;
 
 import javax.xml.namespace.QName;
@@ -50,7 +50,7 @@
  * 2. Intercepts incoming OCPP messages to update the endpoint address ("From" field of the WS-A header) in DB.
  * And the absence of the field is not a deal breaker anymore. But, as a side effect, the user will not be able
  * to send commands to the charging station, since the DB call to list the charge points will filter it out. See
- * {@link ChargePointRepositoryImpl#getChargePointSelect(OcppProtocol, java.util.List)}.
+ * {@link ChargePointRepositoryImpl#getChargePointSelect(OcppProtocol, java.util.List, java.util.List)}.
  *
  * @author Sevket Goekay <[email protected]>
  * @since 15.06.2015
@@ -60,19 +60,19 @@
 public class MessageHeaderInterceptor extends AbstractPhaseInterceptor<Message> {
 
     private final OcppServerRepository ocppServerRepository;
-    private final ChargePointRegistrationService chargePointRegistrationService;
-    private final DelegatingTaskExecutor asyncTaskExecutor;
+    private final ChargePointService chargePointService;
+    private final TaskExecutor taskExecutor;
 
     private static final String BOOT_OPERATION_NAME = "BootNotification";
     private static final String CHARGEBOX_ID_HEADER = "ChargeBoxIdentity";
 
     public MessageHeaderInterceptor(OcppServerRepository ocppServerRepository,
-                                    ChargePointRegistrationService chargePointRegistrationService,
-                                    DelegatingTaskExecutor asyncTaskExecutor) {
+                                    ChargePointService chargePointService,
+                                    TaskExecutor taskExecutor) {
         super(Phase.PRE_INVOKE);
         this.ocppServerRepository = ocppServerRepository;
-        this.chargePointRegistrationService = chargePointRegistrationService;
-        this.asyncTaskExecutor = asyncTaskExecutor;
+        this.chargePointService = chargePointService;
+        this.taskExecutor = taskExecutor;
     }
 
     @Override
@@ -86,7 +86,7 @@ public void handleMessage(Message message) throws Fault {
         QName opName = message.getExchange().getBindingOperationInfo().getOperationInfo().getName();
 
         if (!BOOT_OPERATION_NAME.equals(opName.getLocalPart())) {
-            Optional<RegistrationStatus> status = chargePointRegistrationService.getRegistrationStatus(chargeBoxId);
+            Optional<RegistrationStatus> status = chargePointService.getRegistrationStatus(chargeBoxId);
             boolean allow = status.isPresent() && status.get() != RegistrationStatus.REJECTED;
             if (!allow) {
                 throw createAuthFault(opName);
@@ -97,7 +97,7 @@ public void handleMessage(Message message) throws Fault {
         // 2. update endpoint
         // -------------------------------------------------------------------------
 
-        asyncTaskExecutor.execute(() -> {
+        taskExecutor.execute(() -> {
             try {
                 String endpointAddress = getEndpointAddress(message);
                 if (endpointAddress != null) {

src/main/java/de/rwth/idsg/steve/ocpp/soap/MessageHeaderInterceptor.java

@@ -19,23 +19,21 @@
 package de.rwth.idsg.steve.ocpp.ws;
 
 import com.google.common.base.Strings;
-import de.rwth.idsg.steve.config.DelegatingTaskScheduler;
-import de.rwth.idsg.steve.config.SteveProperties;
 import de.rwth.idsg.steve.config.WebSocketConfiguration;
 import de.rwth.idsg.steve.ocpp.OcppTransport;
 import de.rwth.idsg.steve.ocpp.OcppVersion;
 import de.rwth.idsg.steve.ocpp.ws.data.CommunicationContext;
-import de.rwth.idsg.steve.ocpp.ws.data.SessionContext;
-import de.rwth.idsg.steve.ocpp.ws.pipeline.OcppCallHandler;
 import de.rwth.idsg.steve.ocpp.ws.pipeline.Deserializer;
 import de.rwth.idsg.steve.ocpp.ws.pipeline.IncomingPipeline;
+import de.rwth.idsg.steve.ocpp.ws.pipeline.OcppCallHandler;
 import de.rwth.idsg.steve.repository.OcppServerRepository;
 import de.rwth.idsg.steve.service.notification.OcppStationWebSocketConnected;
 import de.rwth.idsg.steve.service.notification.OcppStationWebSocketDisconnected;
 import org.joda.time.DateTime;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.web.socket.BinaryMessage;
 import org.springframework.web.socket.CloseStatus;
 import org.springframework.web.socket.PongMessage;
@@ -47,9 +45,7 @@
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
-import java.util.Deque;
 import java.util.List;
-import java.util.Map;
 import java.util.concurrent.ScheduledFuture;
 import java.util.function.Consumer;
 
@@ -61,7 +57,7 @@ public abstract class AbstractWebSocketEndpoint extends ConcurrentWebSocketHandl
 
     public static final String CHARGEBOX_ID_KEY = "CHARGEBOX_ID_KEY";
 
-    private final DelegatingTaskScheduler asyncTaskScheduler;
+    private final TaskScheduler taskScheduler;
     private final OcppServerRepository ocppServerRepository;
     private final FutureResponseContextStore futureResponseContextStore;
     private final IncomingPipeline pipeline;
@@ -72,17 +68,17 @@ public abstract class AbstractWebSocketEndpoint extends ConcurrentWebSocketHandl
     private final List<Consumer<String>> disconnectedCallbackList = new ArrayList<>();
     private final Object sessionContextLock = new Object();
 
-    public AbstractWebSocketEndpoint(DelegatingTaskScheduler asyncTaskScheduler,
+    public AbstractWebSocketEndpoint(TaskScheduler taskScheduler,
                                      OcppServerRepository ocppServerRepository,
                                      FutureResponseContextStore futureResponseContextStore,
                                      ApplicationEventPublisher applicationEventPublisher,
-                                     SteveProperties steveProperties,
+                                     SessionContextStoreHolder sessionContextStoreHolder,
                                      AbstractTypeStore typeStore) {
-        this.asyncTaskScheduler = asyncTaskScheduler;
+        this.taskScheduler = taskScheduler;
         this.ocppServerRepository = ocppServerRepository;
         this.futureResponseContextStore = futureResponseContextStore;
         this.pipeline = new IncomingPipeline(new Deserializer(futureResponseContextStore, typeStore), this);
-        this.sessionContextStore = new SessionContextStoreImpl(steveProperties.getOcpp().getWsSessionSelectStrategy());
+        this.sessionContextStore = sessionContextStoreHolder.getOrCreate(getVersion());
 
         connectedCallbackList.add((chargeBoxId) -> applicationEventPublisher.publishEvent(new OcppStationWebSocketConnected(chargeBoxId)));
         disconnectedCallbackList.add((chargeBoxId) -> applicationEventPublisher.publishEvent(new OcppStationWebSocketDisconnected(chargeBoxId)));
@@ -148,7 +144,7 @@ public void onOpen(WebSocketSession session) throws Exception {
 
         // Just to keep the connection alive, such that the servers do not close
         // the connection because of a idle timeout, we ping-pong at fixed intervals.
-        ScheduledFuture pingSchedule = asyncTaskScheduler.scheduleAtFixedRate(
+        ScheduledFuture pingSchedule = taskScheduler.scheduleAtFixedRate(
                 new PingTask(chargeBoxId, session),
                 Instant.now().plus(WebSocketConfiguration.PING_INTERVAL),
                 WebSocketConfiguration.PING_INTERVAL
@@ -210,28 +206,4 @@ protected String getChargeBoxId(WebSocketSession session) {
         return (String) session.getAttributes().get(CHARGEBOX_ID_KEY);
     }
 
-    protected void registerConnectedCallback(Consumer<String> consumer) {
-        connectedCallbackList.add(consumer);
-    }
-
-    protected void registerDisconnectedCallback(Consumer<String> consumer) {
-        disconnectedCallbackList.add(consumer);
-    }
-
-    public List<String> getChargeBoxIdList() {
-        return sessionContextStore.getChargeBoxIdList();
-    }
-
-    public int getNumberOfChargeBoxes() {
-        return sessionContextStore.getNumberOfChargeBoxes();
-    }
-
-    public Map<String, Deque<SessionContext>> getACopy() {
-        return sessionContextStore.getACopy();
-    }
-
-    public WebSocketSession getSession(String chargeBoxId) {
-        return sessionContextStore.getSession(chargeBoxId);
-    }
-
 }