Skip to content

Commit

Permalink
Fixing issues in Event Notification Implementation
Browse files Browse the repository at this point in the history
  • Loading branch information
@Ashi1993
Ashi1993 committed Nov 3, 2023
1 parent 7387034 commit 5e3a107
Show file tree
Hide file tree
Showing 14 changed files with 119 additions and 86 deletions.
2 changes: 1 addition & 1 deletion ...banking/accelerator/event/notifications/service/constants/EventNotificationConstants.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -114,5 +114,5 @@ public class EventNotificationConstants {
public static final String EVENT_SUBSCRIPTION_NOT_FOUND = "Event subscription not found.";
public static final String EVENT_SUBSCRIPTIONS_NOT_FOUND = "Event subscriptions not found for the given client id.";
public static final String ERROR_HANDLING_EVENT_SUBSCRIPTION = "Error occurred while handling the event " +
"subscription request";
"subscription request";
}
40 changes: 21 additions & 19 deletions ...om/wso2/openbanking/accelerator/event/notifications/service/dao/AggregatedPollingDAO.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -35,68 +35,70 @@ public interface AggregatedPollingDAO {
* This method is to update the notification status by ID, allowed values are.
* OPEN,ACK and ERR
*
* @param notificationId
* @param notificationStatus
* @return
* @param notificationId Notification ID to update
* @param notificationStatus Notification status to update
* @return Update is success or not
* @throws OBEventNotificationException
*/
Boolean updateNotificationStatusById(String notificationId, String notificationStatus)
throws OBEventNotificationException;

/**
* This method is to store event notifications in the OB_NOTIFICATION table.
* @param notificationError
* @return
* This method is to store event notifications error details in the OB_NOTIFICATION table.
*
* @param notificationError Notification error details
* @return Stored event notifications error details
* @throws OBEventNotificationException
*/
Map<String, NotificationError> storeErrorNotification(NotificationError notificationError)
throws OBEventNotificationException;

/**
* This method is to retrieve given number of notifications in the OB_NOTIFICATION table by client and status.
* @param clientId
* @param status
* @param max
* @return
*
* @param clientId Client ID to retrieve notifications
* @param status Notification status to retrieve
* @param max Maximum number of notifications to retrieve
* @return List of notifications by client and status
* @throws OBEventNotificationException
*/
List<NotificationDTO> getNotificationsByClientIdAndStatus(String clientId, String
status, int max) throws OBEventNotificationException;

/**
* This method is to retrieve notifications by NotificationID.
* @param notificationId
*
* @return
* @param notificationId Notification ID to retrieve
* @return List of notifications by notification ID
* @throws OBEventNotificationException
*/
List<NotificationEvent> getEventsByNotificationID(String notificationId) throws OBEventNotificationException;

/**
* This method is to retrieve notifications in the OB_NOTIFICATION table by status.
* @param status
*
* @return List<NotificationDTO>
* @param status Notification status to retrieve
* @return List of notifications by status
* @throws OBEventNotificationException
*/
List<NotificationDTO> getNotificationsByStatus(String status) throws OBEventNotificationException;

/**
* This method is to retrieve notificationsCount by ClientId and Status.
* @param clientId
* @param eventStatus
*
* @return
* @param clientId Client ID to retrieve notifications
* @param eventStatus Notification status to retrieve
* @return List of notifications by status and client id
* @throws OBEventNotificationException
*/
int getNotificationCountByClientIdAndStatus(String clientId, String eventStatus)
throws OBEventNotificationException;

/**
* This method is to retrieve the notification status.
* @param notificationId
*
* @return
* @param notificationId Notification ID to retrieve
* @return Notification status by notification ID
* @throws OBEventNotificationException
*/
boolean getNotificationStatus(String notificationId) throws OBEventNotificationException;
Expand Down
76 changes: 40 additions & 36 deletions ...so2/openbanking/accelerator/event/notifications/service/dao/AggregatedPollingDAOImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ public Boolean updateNotificationStatusById(String notificationId, String notifi
Connection connection = DatabaseUtil.getDBConnection();
if (log.isDebugEnabled()) {
log.debug(String.format("Database connection is established for updating notification with " +
"ID : '%s' in the database. ", notificationId));
"ID : '%s' in the database. ", notificationId.replaceAll("[\r\n]", "")));
}
try {
connection.setAutoCommit(false);
Expand All @@ -78,32 +78,31 @@ public Boolean updateNotificationStatusById(String notificationId, String notifi
if (affectedRows != 0) {
connection.commit();
if (log.isDebugEnabled()) {
log.debug("Updated notification with Notification ID : " + notificationId);
log.debug(String.format("Updated notification with Notification ID '%s'",
notificationId.replaceAll("[\r\n]", "")));
}

return true;
} else {
if (log.isDebugEnabled()) {
log.debug("Failed updating notification with ID : " + notificationId);
log.debug(String.format("Failed updating notification with ID : '%s'",
notificationId.replaceAll("[\r\n]", "")));
}
return false;
}
} catch (SQLException e) {
connection.rollback(savepoint);
log.error(String.format(EventNotificationConstants.DB_ERROR_UPDATING, notificationId), e);
log.error(String.format(EventNotificationConstants.DB_ERROR_UPDATING,
notificationId.replaceAll("[\r\n]", "")), e);
throw new OBEventNotificationException(String.format(EventNotificationConstants.DB_ERROR_UPDATING,
notificationId));
}
} catch (SQLException e) {
if (log.isDebugEnabled()) {
log.debug("SQL exception when updating notification status", e);
}
log.debug("SQL exception when updating notification status", e);
throw new OBEventNotificationException("Database error while closing the connection to the" +
" the database.");
} finally {
if (log.isDebugEnabled()) {
log.debug(EventNotificationConstants.DATABASE_CONNECTION_CLOSE_LOG_MSG);
}
log.debug(EventNotificationConstants.DATABASE_CONNECTION_CLOSE_LOG_MSG);
DatabaseUtil.closeConnection(connection);
}
}
Expand All @@ -119,8 +118,9 @@ public Map<String, NotificationError> storeErrorNotification(NotificationError n
connection.setAutoCommit(false);

if (log.isDebugEnabled()) {
log.debug("Database connection is established for storing error notification with ID : "
+ notificationError.getNotificationId());
log.debug(String.format("Database connection is established for storing error notification with ID" +
" : '%s' in the database. ",
notificationError.getNotificationId().replaceAll("[\r\n]", "")));
}

final String storeErrorNotificationQuery = sqlStatements.storeErrorNotificationQuery();
Expand All @@ -138,14 +138,14 @@ public Map<String, NotificationError> storeErrorNotification(NotificationError n
if (affectedRows == 1) {
connection.commit();
if (log.isDebugEnabled()) {
log.debug("Successfully stored error notification with ID : " +
notificationError.getNotificationId());
log.debug(String.format("Successfully stored error notification with ID:'%s'.",
notificationError.getNotificationId().replaceAll("[\r\n]", "")));
}
response.put(notificationError.getNotificationId(), notificationError);
} else {
if (log.isDebugEnabled()) {
log.debug(EventNotificationConstants.DB_FAILED_ERROR_NOTIFICATION_STORING
+ notificationError.getNotificationId());
log.debug(String.format("Failed store error notification with ID:'%s'.",
notificationError.getNotificationId().replaceAll("[\r\n]", "")));
}
throw new OBEventNotificationException(EventNotificationConstants.
DB_FAILED_ERROR_NOTIFICATION_STORING + notificationError.getNotificationId());
Expand Down Expand Up @@ -176,7 +176,8 @@ public List<NotificationDTO> getNotificationsByClientIdAndStatus(String clientId
notificationList = new ArrayList<>();

if (log.isDebugEnabled()) {
log.debug(String.format(EventNotificationConstants.DB_CONN_ESTABLISHED, clientId));
log.debug(String.format(EventNotificationConstants.DB_CONN_ESTABLISHED,
clientId.replaceAll("[\r\n]", "")));
}

final String sql = sqlStatements.getMaxNotificationsQuery();
Expand Down Expand Up @@ -215,12 +216,12 @@ public List<NotificationDTO> getNotificationsByClientIdAndStatus(String clientId

if (log.isDebugEnabled()) {
log.debug(String.format(EventNotificationConstants.RETRIEVED_NOTIFICATION_CLIENT,
clientId));
clientId.replaceAll("[\r\n]", "")));
}
} else {
if (log.isDebugEnabled()) {
log.debug(String.format(EventNotificationConstants.NO_NOTIFICATIONS_FOUND_CLIENT,
clientId));
clientId.replaceAll("[\r\n]", "")));
}
}
}
Expand Down Expand Up @@ -268,29 +269,31 @@ public List<NotificationEvent> getEventsByNotificationID(String notificationId)
(EventNotificationConstants.EVENT_TYPE));
event.setEventInformation(EventNotificationServiceUtil.
getEventJSONFromString(eventsResultSet.getString
(EventNotificationConstants.EVENT_INFO)));
(EventNotificationConstants.EVENT_INFO)));
eventList.add(event);
}
eventsResultSet.close();
getEventsPreparedStatement.close();

if (log.isDebugEnabled()) {
log.debug(String.format(EventNotificationConstants.RETRIEVED_EVENTS_NOTIFICATION,
notificationId));
notificationId.replaceAll("[\r\n]", "")));
}
} else {
if (log.isDebugEnabled()) {
log.debug(String.format(EventNotificationConstants.NO_EVENTS_NOTIFICATION_ID,
notificationId));
notificationId.replaceAll("[\r\n]", "")));
}
}
} catch (ParseException e) {
log.error(String.format(EventNotificationConstants.PARSE_ERROR_NOTIFICATION_ID, notificationId), e);
log.error(String.format(EventNotificationConstants.PARSE_ERROR_NOTIFICATION_ID,
notificationId.replaceAll("[\r\n]", "")), e);
throw new OBEventNotificationException(String.format (
EventNotificationConstants.PARSE_ERROR_NOTIFICATION_ID, notificationId), e);
}
} catch (SQLException e) {
log.error(String.format(EventNotificationConstants.DB_ERROR_EVENTS_RETRIEVE, notificationId), e);
log.error(String.format(EventNotificationConstants.DB_ERROR_EVENTS_RETRIEVE,
notificationId.replaceAll("[\r\n]", "")), e);
throw new OBEventNotificationException(String.format
(EventNotificationConstants.DB_ERROR_EVENTS_RETRIEVE, notificationId), e);
}
Expand Down Expand Up @@ -337,7 +340,8 @@ public List<NotificationDTO> getNotificationsByStatus(String status) throws OBEv
notificationResultSet.close();
getNotificationsPreparedStatement.close();
if (log.isDebugEnabled()) {
log.debug(EventNotificationConstants.RETRIEVED_NOTIFICATION_CLIENT);
log.debug(
EventNotificationConstants.RETRIEVED_NOTIFICATION_CLIENT);
}
} else {
if (log.isDebugEnabled()) {
Expand Down Expand Up @@ -378,24 +382,23 @@ public int getNotificationCountByClientIdAndStatus(String clientId, String event

if (log.isDebugEnabled()) {
log.debug(String.format("Retrieved notification count for client ID: '%s'. ",
clientId));
clientId.replaceAll("[\r\n]", "")));
}

return count;
} else {
if (log.isDebugEnabled()) {
log.debug(String.format(
EventNotificationConstants.NO_NOTIFICATIONS_FOUND_CLIENT,
clientId));
clientId.replaceAll("[\r\n]", "")));
}

return 0;
}
}
} catch (SQLException e) {
throw new OBEventNotificationException(String.format
(EventNotificationConstants.DB_ERROR_NOTIFICATION_RETRIEVE,
clientId), e);
(EventNotificationConstants.DB_ERROR_NOTIFICATION_RETRIEVE, clientId), e);
}
} finally {
log.debug(EventNotificationConstants.DATABASE_CONNECTION_CLOSE_LOG_MSG);
Expand All @@ -422,13 +425,14 @@ public boolean getNotificationStatus(String notificationId) throws OBEventNotifi
isOpenStatus = true;
}

return isOpenStatus;
} else {
if (log.isDebugEnabled()) {
log.debug("No notifications found for notification ID : " + notificationId);
}
}
}
return isOpenStatus;
} else {
if (log.isDebugEnabled()) {
log.debug(String.format("No notifications found for notification ID - '%s'",
notificationId.replaceAll("[\r\n]", "")));
}
}
}
} catch (SQLException e) {
throw new OBEventNotificationException(String.format
("Error occurred while retrieving status for the notifications ID : '%s'.",
Expand Down
6 changes: 3 additions & 3 deletions ...a/com/wso2/openbanking/accelerator/event/notifications/service/dao/EventPublisherDAO.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -32,9 +32,9 @@ public interface EventPublisherDAO {

/**
* This method is used to persist event notifications in the database.
* @param connection
* @param notificationDTO
* @param eventsList
* @param connection Database connection
* @param notificationDTO Notification details DTO
* @param eventsList List of notification events
* @return NotificationID of the saved notification.
* @throws OBEventNotificationException
*/
Expand Down
17 changes: 10 additions & 7 deletions ...so2/openbanking/accelerator/event/notifications/service/dao/PostgreSqlPollingDAOImpl.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -59,7 +59,8 @@ public List<NotificationDTO> getNotificationsByClientIdAndStatus(String clientId
notificationList = new ArrayList<>();

if (log.isDebugEnabled()) {
log.debug(String.format(EventNotificationConstants.DB_CONN_ESTABLISHED, clientId));
log.debug(String.format(EventNotificationConstants.DB_CONN_ESTABLISHED,
clientId.replaceAll("[\r\n]", "")));
}

final String sql = sqlStatements.getMaxNotificationsQuery();
Expand Down Expand Up @@ -99,13 +100,13 @@ public List<NotificationDTO> getNotificationsByClientIdAndStatus(String clientId

if (log.isDebugEnabled()) {
log.debug(String.format(EventNotificationConstants.RETRIEVED_NOTIFICATION_CLIENT,
clientId));
clientId.replaceAll("[\r\n]", "")));
}

} else {
if (log.isDebugEnabled()) {
log.debug(String.format(EventNotificationConstants.NO_NOTIFICATIONS_FOUND_CLIENT,
clientId));
clientId.replaceAll("[\r\n]", "")));
}
}
}
Expand Down Expand Up @@ -162,21 +163,23 @@ public List<NotificationEvent> getEventsByNotificationID(String notificationId)

if (log.isDebugEnabled()) {
log.debug(String.format(EventNotificationConstants.RETRIEVED_EVENTS_NOTIFICATION,
notificationId));
notificationId.replaceAll("[\r\n]", "")));
}
} else {
if (log.isDebugEnabled()) {
log.debug(String.format(EventNotificationConstants.NO_EVENTS_NOTIFICATION_ID,
notificationId));
notificationId.replaceAll("[\r\n]", "")));
}
}
} catch (ParseException e) {
log.error(String.format(EventNotificationConstants.PARSE_ERROR_NOTIFICATION_ID, notificationId), e);
log.error(String.format(EventNotificationConstants.PARSE_ERROR_NOTIFICATION_ID,
notificationId.replaceAll("[\r\n]", "")), e);
throw new OBEventNotificationException(String.format (
EventNotificationConstants.PARSE_ERROR_NOTIFICATION_ID, notificationId), e);
}
} catch (SQLException e) {
log.error(String.format(EventNotificationConstants.DB_ERROR_EVENTS_RETRIEVE, notificationId), e);
log.error(String.format(EventNotificationConstants.DB_ERROR_EVENTS_RETRIEVE,
notificationId.replaceAll("[\r\n]", "")), e);
throw new OBEventNotificationException(String.format
(EventNotificationConstants.DB_ERROR_EVENTS_RETRIEVE, notificationId), e);
}
Expand Down
Loading

0 comments on commit 5e3a107

Please sign in to comment.