1212import org .springframework .stereotype .Component ;
1313
1414import com .mongodb .MongoClient ;
15+ import com .mongodb .MongoClientURI ;
1516import com .mongodb .client .MongoCollection ;
1617import com .mongodb .client .MongoDatabase ;
1718import com .mongodb .client .model .Filters ;
2223
2324@ Component
2425public class DataBaseManager {
25-
26+ private static final int MAX_WAIT_TIME_MILLISECONDS = 30000 ;
27+ private static final int RETRY_EVERY_X_MILLISECONDS = 1000 ;
28+
2629 @ Value ("${spring.data.mongodb.database}" )
2730 private String database ;
2831
2932 @ Value ("${event_object_map.collection.name}" )
30- private String eventMapCollection ;
33+ private String eventObjectCollectionName ;
3134
3235 @ Value ("${aggregated.collection.name}" )
3336 private String aggregatedCollectionName ;
@@ -44,30 +47,22 @@ public class DataBaseManager {
4447
4548 private MongoClient mongoClient ;
4649
47- public int getMongoDbPort () {
48- return mongoProperties .getPort ();
49- }
50-
51- public String getMongoDbHost () {
52- return mongoProperties .getHost ();
53- }
54-
5550 /**
5651 * Verify that aggregated object contains the expected information.
5752 *
58- * @param checklist
59- * list of checklist to check
53+ * @param checklist list of checklist to check
6054 * @return list of missing checklist
6155 * @throws InterruptedException
6256 */
63- public List <String > verifyAggregatedObjectInDB (List <String > checklist ) throws InterruptedException {
64- long stopTime = System .currentTimeMillis () + 30000 ;
57+ public List <String > verifyAggregatedObjectInDB (List <String > checklist )
58+ throws InterruptedException {
59+ long stopTime = System .currentTimeMillis () + MAX_WAIT_TIME_MILLISECONDS ;
6560 while (!checklist .isEmpty () && stopTime > System .currentTimeMillis ()) {
66- checklist = compareArgumentsWithAggregatedObjectInDB (checklist );
61+ checklist = getListOfArgumentsNotFoundInDatabase (checklist );
6762 if (checklist .isEmpty ()) {
6863 break ;
6964 }
70- TimeUnit .MILLISECONDS .sleep (1000 );
65+ TimeUnit .MILLISECONDS .sleep (RETRY_EVERY_X_MILLISECONDS );
7166 }
7267 return checklist ;
7368 }
@@ -79,82 +74,81 @@ public List<String> verifyAggregatedObjectInDB(List<String> checklist) throws In
7974 * @throws InterruptedException
8075 */
8176 public boolean verifyAggregatedObjectExistsInDB () throws InterruptedException {
82- long stopTime = System .currentTimeMillis () + 30000 ;
77+ long stopTime = System .currentTimeMillis () + MAX_WAIT_TIME_MILLISECONDS ;
8378 while (stopTime > System .currentTimeMillis ()) {
84- mongoClient = new MongoClient (getMongoDbHost (), getMongoDbPort ());
85- MongoDatabase db = mongoClient .getDatabase (database );
86- MongoCollection <Document > collection = db .getCollection (aggregatedCollectionName );
87- List <Document > documents = collection .find ().into (new ArrayList <>());
88- TimeUnit .MILLISECONDS .sleep (1000 );
79+ List <Document > documents = getDocumentsFromCollection (aggregatedCollectionName );
8980 if (!documents .isEmpty ()) {
9081 return true ;
9182 }
83+ TimeUnit .MILLISECONDS .sleep (RETRY_EVERY_X_MILLISECONDS );
9284 }
9385 return false ;
9486 }
9587
96- /**
97- * Checks that aggregated object contains specified arguments.
98- *
99- * @param checklist
100- * list of arguments
101- * @return list of missing arguments
102- */
103- private List <String > compareArgumentsWithAggregatedObjectInDB (List <String > checklist ) {
104- mongoClient = new MongoClient (getMongoDbHost (), getMongoDbPort ());
105- MongoDatabase db = mongoClient .getDatabase (database );
106- MongoCollection <Document > collection = db .getCollection (aggregatedCollectionName );
107- List <Document > documents = collection .find ().into (new ArrayList <>());
108- for (Document document : documents ) {
109- for (String expectedValue : new ArrayList <>(checklist )) {
110- if (document .toString ().contains (expectedValue )) {
111- checklist .remove (expectedValue );
112- }
113- }
114- }
115- return checklist ;
116- }
117-
11888 /**
11989 * Verify that events are located in the database collection.
12090 *
121- * @param eventsIdList
122- * list of events IDs
91+ * @param eventsIdList list of events IDs
12392 * @return list of missing events
12493 * @throws InterruptedException
12594 */
126- public List <String > verifyEventsInDB (List <String > eventsIdList , int extraCheckDelay ) throws InterruptedException {
127- long stopTime = System .currentTimeMillis () + 30000 + extraCheckDelay ;
95+ public List <String > verifyEventsInDB (List <String > eventsIdList , int extraCheckDelay )
96+ throws InterruptedException {
97+ long stopTime = System .currentTimeMillis () + MAX_WAIT_TIME_MILLISECONDS + extraCheckDelay ;
12898 while (!eventsIdList .isEmpty () && stopTime > System .currentTimeMillis ()) {
12999 eventsIdList = compareSentEventsWithEventsInDB (eventsIdList );
130100 if (eventsIdList .isEmpty ()) {
131101 break ;
132102 }
133- TimeUnit .MILLISECONDS .sleep (1000 );
103+ TimeUnit .MILLISECONDS .sleep (RETRY_EVERY_X_MILLISECONDS );
134104 }
135105 return eventsIdList ;
136106 }
137107
108+ private List <String > getListOfArgumentsNotFoundInDatabase (final List <String > checklist ) {
109+ final List <Document > documents = getDocumentsFromCollection (aggregatedCollectionName );
110+
111+ List <String > foundValues = new ArrayList <>();
112+ for (Document document : documents ) {
113+ final List <String > valuesFoundInDocument = getValuesFoundInDocument (checklist ,
114+ document );
115+ foundValues .addAll (valuesFoundInDocument );
116+ }
117+
118+ List <String > result = new ArrayList <>(checklist );
119+ result .removeAll (foundValues );
120+ return result ;
121+ }
122+
123+ private List <String > getValuesFoundInDocument (final List <String > checklist ,
124+ final Document document ) {
125+ List <String > foundValues = new ArrayList <>();
126+ for (final String expectedValue : checklist ) {
127+ if (document .toString ().contains (expectedValue )) {
128+ foundValues .add (expectedValue );
129+ }
130+ }
131+ return foundValues ;
132+ }
133+
138134 /**
139135 * Checks collection of events against event list.
140136 *
141- * @param checklist
142- * list of event IDs
137+ * @param checklist list of event IDs
143138 * @return list of missing events
144139 */
145- private List <String > compareSentEventsWithEventsInDB (List <String > checklist ) {
146- mongoClient = new MongoClient (getMongoDbHost (), getMongoDbPort ());
147- MongoDatabase db = mongoClient .getDatabase (database );
148- MongoCollection <Document > collection = db .getCollection (eventMapCollection );
149- List <Document > documents = collection .find ().into (new ArrayList <>());
140+ private List <String > compareSentEventsWithEventsInDB (final List <String > checklist ) {
141+ final List <Document > documents = getDocumentsFromCollection (eventObjectCollectionName );
142+
143+ List <String > foundIDs = new ArrayList <>();
150144 for (Document document : documents ) {
151- for (String expectedID : new ArrayList <>(checklist )) {
152- if (expectedID .equals (document .get ("_id" ).toString ())) {
153- checklist .remove (expectedID );
154- }
155- }
145+ final String documentId = document .get ("_id" ).toString ();
146+ foundIDs .add (documentId );
156147 }
157- return checklist ;
148+
149+ List <String > result = new ArrayList <>(checklist );
150+ result .removeAll (foundIDs );
151+ return result ;
158152 }
159153
160154 /**
@@ -171,31 +165,41 @@ public String getValueFromQuery(List<String> databaseQueryResult, String key, in
171165 return jsonObject .get (key ).toString ();
172166 }
173167
174- /**
175- * Returns the size of the waitlist.
176- *
177- * @return int of the size of the waitlist.
178- */
179- public int waitListSize () {
180- mongoClient = new MongoClient (getMongoDbHost (), getMongoDbPort ());
181- MongoDatabase db = mongoClient .getDatabase (database );
182- MongoCollection <Document > collection = db .getCollection (waitlistCollectionName );
183- List <Document > documents = collection .find ().into (new ArrayList <>());
184- return documents .size ();
185- }
186-
187168 /**
188169 * Get a specific subscription from the database based on the subscription name.
189170 *
190171 * @param subscriptionName
191172 * @return the document as JSON string
192173 */
193174 public String getSubscription (String subscriptionName ) {
194- mongoClient = new MongoClient (getMongoDbHost (), getMongoDbPort ());
195- MongoDatabase db = mongoClient .getDatabase (database );
196- MongoCollection <Document > collection = db .getCollection (subscriptionCollectionName );
175+ MongoCollection <Document > collection = getCollection (subscriptionCollectionName );
197176 Bson filter = Filters .eq ("subscriptionName" , subscriptionName );
198177 Document document = collection .find (filter ).first ();
199178 return document .toJson ();
200179 }
180+
181+ /**
182+ * Returns the size of the waitlist.
183+ *
184+ * @return int of the size of the waitlist.
185+ */
186+ public int waitListSize () {
187+ List <Document > documents = getDocumentsFromCollection (waitlistCollectionName );
188+ return documents .size ();
189+ }
190+
191+ private List <Document > getDocumentsFromCollection (String collectionName ) {
192+ MongoCollection <Document > collection = getCollection (collectionName );
193+ List <Document > documents = collection .find ().into (new ArrayList <>());
194+ return documents ;
195+ }
196+
197+ private MongoCollection <Document > getCollection (String collectionName ) {
198+ MongoClientURI uri = new MongoClientURI (mongoProperties .getUri ());
199+ mongoClient = new MongoClient (uri );
200+ MongoDatabase db = mongoClient .getDatabase (database );
201+ MongoCollection <Document > collection = db .getCollection (collectionName );
202+ return collection ;
203+ }
204+
201205}
0 commit comments