Fixed broken data consumption logic due to missing bundleId in OcsEvent

ostelco · Aug 22, 2018 · d9c3a0c · d9c3a0c
2 parents b642713 + a63442a
commit d9c3a0c
Show file tree

Hide file tree

Showing 24 changed files with 88 additions and 80 deletions.
diff --git a/README.md b/README.md
@@ -1,5 +1,5 @@
 
-[![Kotlin version badge](https://img.shields.io/badge/kotlin-1.2.60-blue.svg)](http://kotlinlang.org/)
+[![Kotlin version badge](https://img.shields.io/badge/kotlin-1.2.61-blue.svg)](http://kotlinlang.org/)
 [![Prime version](https://img.shields.io/github/tag/ostelco/ostelco-core.svg)](https://github.com/ostelco/ostelco-core/tags)
 [![GitHub license](https://img.shields.io/github/license/ostelco/ostelco-core.svg)](https://github.com/ostelco/ostelco-core/blob/master/LICENSE)
 

diff --git a/acceptance-tests/build.gradle b/acceptance-tests/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "application"
   id "com.github.johnrengelman.shadow" version "2.0.4"
 }

diff --git a/acceptance-tests/src/main/kotlin/org/ostelco/at/pgw/OcsTest.kt b/acceptance-tests/src/main/kotlin/org/ostelco/at/pgw/OcsTest.kt
@@ -1,7 +1,6 @@
 package org.ostelco.at.pgw
 
 import org.jdiameter.api.Avp
-import org.jdiameter.api.AvpDataException
 import org.jdiameter.api.Session
 import org.junit.After
 import org.junit.Before
@@ -11,9 +10,12 @@ import org.ostelco.at.common.createProfile
 import org.ostelco.at.common.createSubscription
 import org.ostelco.at.common.logger
 import org.ostelco.at.common.randomInt
+import org.ostelco.at.jersey.get
 import org.ostelco.diameter.model.RequestType
 import org.ostelco.diameter.test.TestClient
 import org.ostelco.diameter.test.TestHelper
+import org.ostelco.prime.client.model.SubscriptionStatus
+import java.lang.Thread.sleep
 import kotlin.test.assertEquals
 import kotlin.test.fail
 
@@ -57,22 +59,17 @@ class OcsTest {
 
         waitForAnswer()
 
-        try {
-            assertEquals(2001L, client.resultCodeAvp?.integer32?.toLong())
-            val resultAvps = client.resultAvps ?: fail("Missing AVPs")
-            assertEquals(RequestType.INITIAL_REQUEST.toLong(), resultAvps.getAvp(Avp.CC_REQUEST_TYPE).integer32.toLong())
-            assertEquals(DEST_HOST, resultAvps.getAvp(Avp.ORIGIN_HOST).utF8String)
-            assertEquals(DEST_REALM, resultAvps.getAvp(Avp.ORIGIN_REALM).utF8String)
-            val resultMSCC = resultAvps.getAvp(Avp.MULTIPLE_SERVICES_CREDIT_CONTROL)
-            assertEquals(2001L, resultMSCC.grouped.getAvp(Avp.RESULT_CODE).integer32.toLong())
-            assertEquals(1, resultMSCC.grouped.getAvp(Avp.SERVICE_IDENTIFIER_CCA).unsigned32)
-            assertEquals(10, resultMSCC.grouped.getAvp(Avp.RATING_GROUP).unsigned32)
-            val granted = resultMSCC.grouped.getAvp(Avp.GRANTED_SERVICE_UNIT)
-            assertEquals(BUCKET_SIZE, granted.grouped.getAvp(Avp.CC_TOTAL_OCTETS).unsigned64)
-        } catch (e: AvpDataException) {
-            logger.error("Failed to get Result-Code", e)
-        }
-
+        assertEquals(2001L, client.resultCodeAvp?.integer32?.toLong())
+        val resultAvps = client.resultAvps ?: fail("Missing AVPs")
+        assertEquals(RequestType.INITIAL_REQUEST.toLong(), resultAvps.getAvp(Avp.CC_REQUEST_TYPE).integer32.toLong())
+        assertEquals(DEST_HOST, resultAvps.getAvp(Avp.ORIGIN_HOST).utF8String)
+        assertEquals(DEST_REALM, resultAvps.getAvp(Avp.ORIGIN_REALM).utF8String)
+        val resultMSCC = resultAvps.getAvp(Avp.MULTIPLE_SERVICES_CREDIT_CONTROL)
+        assertEquals(2001L, resultMSCC.grouped.getAvp(Avp.RESULT_CODE).integer32.toLong())
+        assertEquals(1, resultMSCC.grouped.getAvp(Avp.SERVICE_IDENTIFIER_CCA).unsigned32)
+        assertEquals(10, resultMSCC.grouped.getAvp(Avp.RATING_GROUP).unsigned32)
+        val granted = resultMSCC.grouped.getAvp(Avp.GRANTED_SERVICE_UNIT)
+        assertEquals(BUCKET_SIZE, granted.grouped.getAvp(Avp.CC_TOTAL_OCTETS).unsigned64)
     }
 
     private fun simpleCreditControlRequestUpdate(session: Session) {
@@ -91,20 +88,25 @@ class OcsTest {
 
         waitForAnswer()
 
-        try {
-            assertEquals(2001L, client.resultCodeAvp?.integer32?.toLong())
-            val resultAvps = client.resultAvps ?: fail("Missing AVPs")
-            assertEquals(DEST_HOST, resultAvps.getAvp(Avp.ORIGIN_HOST).utF8String)
-            assertEquals(DEST_REALM, resultAvps.getAvp(Avp.ORIGIN_REALM).utF8String)
-            assertEquals(RequestType.UPDATE_REQUEST.toLong(), resultAvps.getAvp(Avp.CC_REQUEST_TYPE).integer32.toLong())
-            val resultMSCC = resultAvps.getAvp(Avp.MULTIPLE_SERVICES_CREDIT_CONTROL)
-            assertEquals(2001L, resultMSCC.grouped.getAvp(Avp.RESULT_CODE).integer32.toLong())
-            val granted = resultMSCC.grouped.getAvp(Avp.GRANTED_SERVICE_UNIT)
-            assertEquals(BUCKET_SIZE, granted.grouped.getAvp(Avp.CC_TOTAL_OCTETS).unsigned64)
-        } catch (e: AvpDataException) {
-            logger.error("Failed to get Result-Code", e)
-        }
+        assertEquals(2001L, client.resultCodeAvp?.integer32?.toLong())
+        val resultAvps = client.resultAvps ?: fail("Missing AVPs")
+        assertEquals(DEST_HOST, resultAvps.getAvp(Avp.ORIGIN_HOST).utF8String)
+        assertEquals(DEST_REALM, resultAvps.getAvp(Avp.ORIGIN_REALM).utF8String)
+        assertEquals(RequestType.UPDATE_REQUEST.toLong(), resultAvps.getAvp(Avp.CC_REQUEST_TYPE).integer32.toLong())
+        val resultMSCC = resultAvps.getAvp(Avp.MULTIPLE_SERVICES_CREDIT_CONTROL)
+        assertEquals(2001L, resultMSCC.grouped.getAvp(Avp.RESULT_CODE).integer32.toLong())
+        val granted = resultMSCC.grouped.getAvp(Avp.GRANTED_SERVICE_UNIT)
+        assertEquals(BUCKET_SIZE, granted.grouped.getAvp(Avp.CC_TOTAL_OCTETS).unsigned64)
+    }
+
+    private fun getBalance(): Long {
+        sleep(200) // wait for 200 ms for balance to be updated in db
 
+        val subscriptionStatus: SubscriptionStatus = get {
+            path = "/subscription/status"
+            subscriberId = EMAIL
+        }
+        return subscriptionStatus.remaining
     }
 
     @Test
@@ -114,7 +116,10 @@ class OcsTest {
 
         val session = client.createSession() ?: fail("Failed to create session")
         simpleCreditControlRequestInit(session)
+        assertEquals(INITIAL_BALANCE - BUCKET_SIZE, getBalance(), message = "Incorrect balance after init")
+
         simpleCreditControlRequestUpdate(session)
+        assertEquals(INITIAL_BALANCE - 2 * BUCKET_SIZE, getBalance(), message = "Incorrect balance after update")
 
         val request = client.createRequest(
                 DEST_REALM,
@@ -128,22 +133,19 @@ class OcsTest {
 
         waitForAnswer()
 
-        try {
-            assertEquals(2001L, client.resultCodeAvp?.integer32?.toLong())
-            val resultAvps = client.resultAvps ?: fail("Missing AVPs")
-            assertEquals(DEST_HOST, resultAvps.getAvp(Avp.ORIGIN_HOST).utF8String)
-            assertEquals(DEST_REALM, resultAvps.getAvp(Avp.ORIGIN_REALM).utF8String)
-            assertEquals(RequestType.TERMINATION_REQUEST.toLong(), resultAvps.getAvp(Avp.CC_REQUEST_TYPE).integer32.toLong())
-            val resultMSCC = resultAvps.getAvp(Avp.MULTIPLE_SERVICES_CREDIT_CONTROL)
-            assertEquals(2001L, resultMSCC.grouped.getAvp(Avp.RESULT_CODE).integer32.toLong())
-            assertEquals(1, resultMSCC.grouped.getAvp(Avp.SERVICE_IDENTIFIER_CCA).unsigned32)
-            assertEquals(10, resultMSCC.grouped.getAvp(Avp.RATING_GROUP).unsigned32)
-            val validTime = resultMSCC.grouped.getAvp(Avp.VALIDITY_TIME)
-            assertEquals(86400L, validTime.unsigned32)
-        } catch (e: AvpDataException) {
-            logger.error("Failed to get Result-Code", e)
-        }
-
+        assertEquals(2001L, client.resultCodeAvp?.integer32?.toLong())
+        val resultAvps = client.resultAvps ?: fail("Missing AVPs")
+        assertEquals(DEST_HOST, resultAvps.getAvp(Avp.ORIGIN_HOST).utF8String)
+        assertEquals(DEST_REALM, resultAvps.getAvp(Avp.ORIGIN_REALM).utF8String)
+        assertEquals(RequestType.TERMINATION_REQUEST.toLong(), resultAvps.getAvp(Avp.CC_REQUEST_TYPE).integer32.toLong())
+        val resultMSCC = resultAvps.getAvp(Avp.MULTIPLE_SERVICES_CREDIT_CONTROL)
+        assertEquals(2001L, resultMSCC.grouped.getAvp(Avp.RESULT_CODE).integer32.toLong())
+        assertEquals(1, resultMSCC.grouped.getAvp(Avp.SERVICE_IDENTIFIER_CCA).unsigned32)
+        assertEquals(10, resultMSCC.grouped.getAvp(Avp.RATING_GROUP).unsigned32)
+        val validTime = resultMSCC.grouped.getAvp(Avp.VALIDITY_TIME)
+        assertEquals(86400L, validTime.unsigned32)
+
+        assertEquals(INITIAL_BALANCE - 2 * BUCKET_SIZE, getBalance(), message = "Incorrect balance after terminate")
     }
 
 
@@ -165,7 +167,7 @@ class OcsTest {
 
         waitForAnswer()
 
-        try {
+        run {
             assertEquals(2001L, client.resultCodeAvp?.integer32?.toLong())
             val resultAvps = client.resultAvps ?: fail("Missing AVPs")
             assertEquals(DEST_HOST, resultAvps.getAvp(Avp.ORIGIN_HOST).utF8String)
@@ -176,12 +178,11 @@ class OcsTest {
             assertEquals(1, resultMSCC.grouped.getAvp(Avp.SERVICE_IDENTIFIER_CCA).integer32.toLong())
             val granted = resultMSCC.grouped.getAvp(Avp.GRANTED_SERVICE_UNIT)
             assertEquals(0L, granted.grouped.getAvp(Avp.CC_TOTAL_OCTETS).unsigned64)
-        } catch (e: AvpDataException) {
-            logger.error("Failed to get Result-Code", e)
         }
-
         // There is 2 step in graceful shutdown. First OCS send terminate, then P-GW report used units in a final update
 
+        assertEquals(INITIAL_BALANCE, getBalance(), message = "Incorrect balance after init using wrong msisdn")
+
         val updateRequest = client.createRequest(
                 DEST_REALM,
                 DEST_HOST,
@@ -194,7 +195,7 @@ class OcsTest {
 
         waitForAnswer()
 
-        try {
+        run {
             assertEquals(2001L, client.resultCodeAvp?.integer32?.toLong())
             val resultAvps = client.resultAvps ?: fail("Missing AVPs")
             assertEquals(DEST_HOST, resultAvps.getAvp(Avp.ORIGIN_HOST).utF8String)
@@ -205,10 +206,9 @@ class OcsTest {
             assertEquals(1, resultMSCC.grouped.getAvp(Avp.SERVICE_IDENTIFIER_CCA).integer32.toLong())
             val validTime = resultMSCC.grouped.getAvp(Avp.VALIDITY_TIME)
             assertEquals(86400L, validTime.unsigned32)
-        } catch (e: AvpDataException) {
-            logger.error("Failed to get Result-Code", e)
         }
 
+        assertEquals(INITIAL_BALANCE, getBalance(), message = "Incorrect balance after update using wrong msisdn")
     }
 
 
@@ -234,18 +234,20 @@ class OcsTest {
         private const val DEST_REALM = "loltel"
         private const val DEST_HOST = "ocs"
 
+        private const val INITIAL_BALANCE = 100_000_000L
         private const val BUCKET_SIZE = 500L
 
+        private lateinit var EMAIL: String
         private lateinit var MSISDN: String
 
         @BeforeClass
         @JvmStatic
         fun createTestUserAndSubscription() {
 
-            val email = "ocs-${randomInt()}@test.com"
-            createProfile(name = "Test OCS User", email = email)
+            EMAIL = "ocs-${randomInt()}@test.com"
+            createProfile(name = "Test OCS User", email = EMAIL)
 
-            MSISDN = createSubscription(email)
+            MSISDN = createSubscription(EMAIL)
         }
     }
 }
diff --git a/admin-api/build.gradle b/admin-api/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "java-library"
 }
 

diff --git a/analytics/build.gradle b/analytics/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "application"
   id "com.github.johnrengelman.shadow" version "2.0.4"
   id "idea"

diff --git a/app-notifier/build.gradle b/app-notifier/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "java-library"
 }
 

diff --git a/auth-server/build.gradle b/auth-server/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "application"
   id "com.github.johnrengelman.shadow" version "2.0.4"
   id "idea"

diff --git a/build.gradle b/build.gradle
@@ -28,7 +28,7 @@ subprojects {
     options.encoding = 'UTF-8'
   }
   ext {
-    kotlinVersion = "1.2.60"
+    kotlinVersion = "1.2.61"
     dropwizardVersion = "1.3.5"
     googleCloudVersion = "1.35.0"
     jacksonVersion = "2.9.6"

diff --git a/client-api/build.gradle b/client-api/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-    id "org.jetbrains.kotlin.jvm" version "1.2.60"
+    id "org.jetbrains.kotlin.jvm" version "1.2.61"
     id "java-library"
 }
 

diff --git a/diameter-stack/build.gradle b/diameter-stack/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "java-library"
   id "signing"
   id "maven"

diff --git a/diameter-test/build.gradle b/diameter-test/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "java-library"
   id "signing"
   id "maven"

diff --git a/ext-auth-provider/build.gradle b/ext-auth-provider/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "application"
   id "com.github.johnrengelman.shadow" version "2.0.4"
 }

diff --git a/firebase-store/build.gradle b/firebase-store/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "java-library"
 }
 

diff --git a/model/build.gradle b/model/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "java-library"
 }
 

diff --git a/neo4j-store/build.gradle b/neo4j-store/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "java-library"
 }
 

diff --git a/ocs/build.gradle b/ocs/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "java-library"
 }
 

diff --git a/ocs/src/main/kotlin/org/ostelco/prime/events/EventProcessor.kt b/ocs/src/main/kotlin/org/ostelco/prime/events/EventProcessor.kt
@@ -3,6 +3,7 @@ package org.ostelco.prime.events
 import com.lmax.disruptor.EventHandler
 import org.ostelco.prime.disruptor.EventMessageType.CREDIT_CONTROL_REQUEST
 import org.ostelco.prime.disruptor.EventMessageType.RELEASE_RESERVED_BUCKET
+import org.ostelco.prime.disruptor.EventMessageType.REMOVE_MSISDN_TO_BUNDLE_MAPPING
 import org.ostelco.prime.disruptor.EventMessageType.TOPUP_DATA_BUNDLE_BALANCE
 import org.ostelco.prime.disruptor.OcsEvent
 import org.ostelco.prime.logger
@@ -28,7 +29,8 @@ class EventProcessor(
         try {
             if (event.messageType == CREDIT_CONTROL_REQUEST
                     || event.messageType == RELEASE_RESERVED_BUCKET
-                    || event.messageType == TOPUP_DATA_BUNDLE_BALANCE) {
+                    || event.messageType == TOPUP_DATA_BUNDLE_BALANCE
+                    || event.messageType == REMOVE_MSISDN_TO_BUNDLE_MAPPING) {
                 logger.info("Updating data bundle balance for {} : {} to {} bytes",
                         event.msisdn, event.bundleId, event.bundleBytes)
                 val bundleId = event.bundleId

diff --git a/ocs/src/main/kotlin/org/ostelco/prime/ocs/OcsState.kt b/ocs/src/main/kotlin/org/ostelco/prime/ocs/OcsState.kt
@@ -45,7 +45,8 @@ class OcsState(val loadSubscriberInfo:Boolean = true) : EventHandler<OcsEvent> {
                     event.reservedBucketBytes = reserveDataBytes(
                             msisdn,
                             event.requestedBucketBytes)
-                    event.bundleBytes = getDataBundleBytes(msisdn = msisdn)
+                    event.bundleId = msisdnToBundleIdMap[msisdn]
+                    event.bundleBytes = bundleBalanceMap[event.bundleId] ?: 0
                 }
                 TOPUP_DATA_BUNDLE_BALANCE -> {
                     val bundleId = event.bundleId
@@ -63,6 +64,8 @@ class OcsState(val loadSubscriberInfo:Boolean = true) : EventHandler<OcsEvent> {
                         return
                     }
                     releaseReservedBucket(msisdn = msisdn)
+                    event.bundleId = msisdnToBundleIdMap[msisdn]
+                    event.bundleBytes = bundleBalanceMap[event.bundleId] ?: 0
                 }
                 UPDATE_BUNDLE -> {
                     val bundleId = event.bundleId
@@ -98,9 +101,10 @@ class OcsState(val loadSubscriberInfo:Boolean = true) : EventHandler<OcsEvent> {
                         logger.error("Received null as bundleId")
                         return
                     }
+                    releaseReservedBucket(msisdn = msisdn)
+                    event.bundleBytes = bundleBalanceMap[bundleId] ?: 0
                     msisdnToBundleIdMap.remove(msisdn)
                     bundleIdToMsisdnMap[bundleId]?.remove(msisdn)
-                    // TODO vihang: return reserved bytes back to bundle
                 }
             }
         } catch (e: Exception) {

diff --git a/payment-processor/build.gradle b/payment-processor/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "java-library"
   id "idea"
 }

diff --git a/prime-api/build.gradle b/prime-api/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "java-library"
 }
 

diff --git a/prime-client-api/build.gradle b/prime-client-api/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id 'java-library'
   id 'org.hidetake.swagger.generator' version '2.12.0'
   id "idea"

diff --git a/prime/build.gradle b/prime/build.gradle
@@ -1,5 +1,5 @@
 plugins {
-  id "org.jetbrains.kotlin.jvm" version "1.2.60"
+  id "org.jetbrains.kotlin.jvm" version "1.2.61"
   id "application"
   id "com.github.johnrengelman.shadow" version "2.0.4"
   id "idea"