-
Notifications
You must be signed in to change notification settings - Fork 34
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1176 from OSGP/feature/throttling_db_events
Implement pool for high prio requests based on db events
- Loading branch information
Showing
9 changed files
with
521 additions
and
24 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
45 changes: 45 additions & 0 deletions
45
...g-service/src/main/java/org/opensmartgridplatform/throttling/config/ThrottlingConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,45 @@ | ||
| // SPDX-FileCopyrightText: Copyright Contributors to the GXF project | ||
| // | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package org.opensmartgridplatform.throttling.config; | ||
|
|
||
| import com.zaxxer.hikari.util.DriverDataSource; | ||
| import java.util.Properties; | ||
| import org.opensmartgridplatform.throttling.service.PermitReleasedNotifier; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import org.springframework.beans.factory.annotation.Value; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.Configuration; | ||
|
|
||
| @Configuration | ||
| public class ThrottlingConfig { | ||
| private static final Logger LOGGER = LoggerFactory.getLogger(ThrottlingConfig.class); | ||
|
|
||
| @Value("${spring.datasource.url}") | ||
| private String jdbcUrl; | ||
|
|
||
| @Value("${spring.datasource.driver-class-name}") | ||
| private String databaseDriver; | ||
|
|
||
| @Value("${spring.datasource.username}") | ||
| private String databaseUsername; | ||
|
|
||
| @Value("${spring.datasource.password}") | ||
| private String databasePassword; | ||
|
|
||
| @Bean | ||
| public PermitReleasedNotifier permitReleasedNotifier() { | ||
| LOGGER.info("Created jdbcUrl {} for permitReleasedNotifier", this.jdbcUrl); | ||
| final DriverDataSource dataSource = | ||
| new DriverDataSource( | ||
| this.jdbcUrl, | ||
| this.databaseDriver, | ||
| new Properties(), | ||
| this.databaseUsername, | ||
| this.databasePassword); | ||
|
|
||
| return new PermitReleasedNotifier(dataSource); | ||
| } | ||
| } |
112 changes: 112 additions & 0 deletions
112
...ce/src/main/java/org/opensmartgridplatform/throttling/service/PermitReleasedNotifier.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,112 @@ | ||
| // SPDX-FileCopyrightText: Copyright Contributors to the GXF project | ||
| // | ||
| // SPDX-License-Identifier: Apache-2.0 | ||
|
|
||
| package org.opensmartgridplatform.throttling.service; | ||
|
|
||
| import java.sql.Connection; | ||
| import java.sql.PreparedStatement; | ||
| import java.sql.SQLException; | ||
| import javax.sql.DataSource; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.postgresql.PGConnection; | ||
| import org.postgresql.PGNotification; | ||
|
|
||
| @Slf4j | ||
| public class PermitReleasedNotifier { | ||
| private static final String PERMIT_RELEASED_CHANNEL = "permit_released"; | ||
| private final DataSource dataSource; | ||
|
|
||
| public PermitReleasedNotifier(final DataSource dataSource) { | ||
| this.dataSource = dataSource; | ||
| } | ||
|
|
||
| public boolean waitForAvailablePermit( | ||
| final int baseTransceiverStationId, final int cellId, final int maxWaitInMs) { | ||
| Connection connection = null; | ||
| try { | ||
| connection = this.dataSource.getConnection(); | ||
| final PGConnection pgConnection = | ||
| this.startListening(connection, baseTransceiverStationId, cellId); | ||
|
|
||
| final PGNotification[] notifications = pgConnection.getNotifications(maxWaitInMs); | ||
| if (notifications == null) { | ||
| log.info( | ||
| "Received no notifications for btsId: {}, cellId: {} within {} ms", | ||
| baseTransceiverStationId, | ||
| cellId, | ||
| maxWaitInMs); | ||
| return false; | ||
| } | ||
|
|
||
| log.debug( | ||
| "Received notification for btsId: {}, cellId: {}", baseTransceiverStationId, cellId); | ||
| } catch (final SQLException sqle) { | ||
| log.error( | ||
| "SQLException occurred while listening for notification for btsId: {}, cellId: {}", | ||
| baseTransceiverStationId, | ||
| cellId, | ||
| sqle); | ||
| return false; | ||
| } finally { | ||
| this.closeConnection(connection, baseTransceiverStationId, cellId); | ||
| } | ||
|
|
||
| return true; | ||
| } | ||
|
|
||
| public void notifyPermitReleased(final int baseTransceiverStationId, final int cellId) { | ||
| Connection connection = null; | ||
| try { | ||
| connection = this.dataSource.getConnection(); | ||
|
|
||
| final String sqlStatement = | ||
| String.format("NOTIFY %s", this.getChannelName(baseTransceiverStationId, cellId)); | ||
| try (final PreparedStatement preparedStatement = connection.prepareStatement(sqlStatement)) { | ||
| preparedStatement.executeUpdate(); | ||
| } | ||
| } catch (final SQLException sqle) { | ||
| log.error( | ||
| "SQLException occurred while notify for btsId: {}, cellId: {}", | ||
| baseTransceiverStationId, | ||
| cellId, | ||
| sqle); | ||
| } finally { | ||
| this.closeConnection(connection, baseTransceiverStationId, cellId); | ||
| } | ||
| } | ||
|
|
||
| private PGConnection startListening( | ||
| final Connection connection, final int baseTransceiverStationId, final int cellId) | ||
| throws SQLException { | ||
|
|
||
| final PGConnection pgConnection = connection.unwrap(PGConnection.class); | ||
| final String sqlStatement = | ||
| String.format("LISTEN %s", this.getChannelName(baseTransceiverStationId, cellId)); | ||
| try (final PreparedStatement preparedStatement = connection.prepareStatement(sqlStatement)) { | ||
| preparedStatement.executeUpdate(); | ||
| } | ||
|
|
||
| return pgConnection; | ||
| } | ||
|
|
||
| private void closeConnection( | ||
| final Connection connection, final int baseTransceiverStationId, final int cellId) { | ||
| try { | ||
| if (connection != null) { | ||
| connection.close(); | ||
| } | ||
| } catch (final SQLException e) { | ||
| log.error( | ||
| "SQLException occurred while trying to close the connection that is listening for notification for btsId: {}, cellId: {}", | ||
| baseTransceiverStationId, | ||
| cellId, | ||
| e); | ||
| } | ||
| } | ||
|
|
||
| private String getChannelName(final int baseTransceiverStationId, final int cellId) { | ||
| return (PERMIT_RELEASED_CHANNEL + "_" + baseTransceiverStationId + "_" + cellId) | ||
| .replace("-", "minus"); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.