Changes done for v2.3.3.

Author: nysenthil

com.ibm.streamsx.sttgateway/CHANGELOG.md

@@ -1,5 +1,14 @@
 # Changes
 
+## v2.3.3
+* May/10/2022
+* Added code and logic necessary to scale the VgwDataRouter to process more voice calls with reduced use of CPU cycles.
+* Filtered empty speech packets arrving from either SBC or Voice Gateway.
+* Added more metrics in the VoiceGatewaySource operator to be queryable via the Curl command.
+* Added a hostpool to place the speech processor jobs based on a custom host tag.
+* Activated the option to consider a single EOCS to be treated as the end of a voice call instead of two EOCS signals.
+* Added a feature to change VgwSessionLogging in the VoiceGatewaySource operator via the Curl command.
+
 ## v2.3.2
 * Jan/10/2022
 * Fixed a problem where the call start date time values were not always correctly included in the STT result.

com.ibm.streamsx.sttgateway/com.ibm.streamsx.sttgateway.watson/IBMVoiceGatewaySource/IBMVoiceGatewaySource.xml

@@ -135,6 +135,10 @@
           <function>
             <description>Returns an rstring value indicating the IBM Voice Gateway session id that corresponds to the current output tuple.</description>
             <prototype><![CDATA[rstring getIBMVoiceGatewaySessionId()]]></prototype>
+          </function> 
+          <function>
+            <description>Returns an integer value indicating the unique sequence number of a voice call.</description>
+            <prototype><![CDATA[int32 getCallSequenceNumber()]]></prototype>
           </function>                    
           <function>
             <description>Returns a boolean value to indicate if this is a customer's speech data or not.</description>

com.ibm.streamsx.sttgateway/com.ibm.streamsx.sttgateway.watson/IBMVoiceGatewaySource/IBMVoiceGatewaySource_cpp.cgt

@@ -1,14 +1,14 @@
 /*
 ==============================================
 # Licensed Materials - Property of IBM
-# Copyright IBM Corp. 2019, 2021
+# Copyright IBM Corp. 2019, 2022
 ==============================================
 */
 
 /*
 ============================================================
 First created on: Sep/17/2019
-Last modified on: Sep/17/2021
+Last modified on: May/10/2022
 ============================================================
 */
 
@@ -58,6 +58,8 @@ public:
 	SPL::int64 vgwStaleSessionPurgeInterval;
 	SPL::uint32 maxConcurrentCallsAllowed;
 	SPL::uint32 activeConcurrentCallsCnt;
+	SPL::uint32 peakConcurrentCallsCnt;
+	SPL::uint32 emptySpeechPacketsCnt;
 	bool ipv6Available;
 	server_plain endpoint_plain;
 	server_tls endpoint_tls;
@@ -155,6 +157,20 @@ public:
 	// EndOfCall signal.
 	std::map<std::string, SPL::list<int64_t>> vgw_session_id_map; 
 
+	// This map's key is vgwSessionId and value is 
+	// a call sequence number of type int32_t.
+	// This map stores the unique call sequence number generated and
+	// assigned by this operator to every call.
+	// This unique call sequence number can be used by the 
+	// downstream operators for whatever need they may have for it.
+	// e-g: partition the voice call data based on this unique 
+	// number to distribute in a parallel region to achieve
+	// even load distribution across all the parallel channels.
+	std::map<std::string, int32_t> call_sequence_number_map; 
+	
+	// This counter is used to keep the call sequence number.
+	int32_t callSequenceNumber;
+	
 	// Custom metrics for this operator.
 	Metric *nVoiceCallsProcessedMetric;
 	Metric *nVoiceCallsThrottledMetric;
@@ -235,6 +251,7 @@ public:
 private:
 	// These are the output attribute assignment functions for this operator.
 	std::string getIBMVoiceGatewaySessionId(std::string const & vgwSessionId);
+	int32_t getCallSequenceNumber(int32_t const & callSequenceNumber);
 	bool isCustomerSpeechData(bool const & isThisSpeechDataFromCustomer);
 	int32_t getTupleCnt(int32_t const & emittedTupleCnt);
 	int32_t getTotalSpeechDataBytesReceived(int32_t const & totalSpeechDataBytesReceived);

com.ibm.streamsx.sttgateway/com.ibm.streamsx.sttgateway.watson/IBMVoiceGatewaySource/IBMVoiceGatewaySource_h.cgt

@@ -14,7 +14,7 @@
 
     **Note:** This toolkit requires c++11 support.
     </description>
-    <version>2.3.2</version>
+    <version>2.3.3</version>
     <requiredProductVersion>4.2.1.6</requiredProductVersion>
   </identity>
   <dependencies>

com.ibm.streamsx.sttgateway/info.xml

@@ -1,14 +1,14 @@
 /*
 ==============================================
 # Licensed Materials - Property of IBM
-# Copyright IBM Corp. 2018, 2021
+# Copyright IBM Corp. 2018, 2022
 ==============================================
 */
 
 /*
 ==============================================
 First created on: Nov/24/2020
-Last modified on: Sep/10/2021
+Last modified on: May/10/2022
 
 This is a utility composite that will get used in the following applications.
 
@@ -29,7 +29,8 @@ namespace com.ibm.streamsx.sttgateway.utils;
 // very important and the other ones are purely optional if some
 // scenarios really require them.
 // blob speech --> Speech fragments of a live conversation as captured and sent by the IBM Voice Gateway.
-// rstring vgwSessionId --> Unique identifier of a voice call. 
+// rstring vgwSessionId --> Unique identifier of a voice call.
+// int32 callSequenceNumber --> Unique sequence number for a call. 
 // boolean isCustomerSpeechData --> Every voice call will have a customer channel and an agent channel.
 //                                  This attribute tells whether this output stream carries customer speech data or not.
 // int32 vgwVoiceChannelNumber --> This indicates the voice channel number i.e. 1 or 2.
@@ -51,14 +52,16 @@ namespace com.ibm.streamsx.sttgateway.utils;
 // int32 int32 speechProcessorId --> Id of the speech processor job that is analyzing the speech data.
 // int32 speechEngineId --> This attribute will be set in the speech processor job. (Please, read the comments there.)
 // int32 speechResultProcessorId --> This attribute will be set in the speech processor job. (Please, read the comments there.)
-type BinarySpeech_t = blob speech, rstring vgwSessionId, boolean isCustomerSpeechData, 
+// int32 parallelChannel --> It indicates the parallel channel to be used in the vgwDataRouter job.
+type BinarySpeech_t = blob speech, rstring vgwSessionId, int32 callSequenceNumber, boolean isCustomerSpeechData, 
 	int32 vgwVoiceChannelNumber, boolean endOfCallSignal, 
 	rstring id, rstring callStartDateTime,
 	int64 callStartTimeInEpochSeconds, 
 	rstring callerPhoneNumber, rstring agentPhoneNumber, 
 	rstring ciscoGuid, int32 speechDataFragmentCnt, 
 	int32 totalSpeechDataBytesReceived, int32 speechProcessorId, 
-	int32 speechEngineId, int32 speechResultProcessorId;
+	int32 speechEngineId, int32 speechResultProcessorId,
+	int32 parallelChannel;
 
 // The following schema will be for the data being sent here by the
 // VgwDataRouter application. It sends us raw binary data which 
@@ -70,7 +73,7 @@ type DataFromVgwRouter_t = int32 msgType, blob payload;
 // The following schema is for the call recording feature where we will
 // store the call metadata details for a specific voice channel of a given
 // voice call.
-type CallMetaData_t = rstring vgwSessionId, boolean isCustomerSpeechData, 
+type CallMetaData_t = rstring vgwSessionId, int32 callSequenceNumber, boolean isCustomerSpeechData, 
 	int32 vgwVoiceChannelNumber, rstring callerPhoneNumber, 
 	rstring agentPhoneNumber, rstring ciscoGuid;
 
@@ -110,7 +113,8 @@ type SendData_t = rstring strData, blob blobData;
 
 // The following schema defines the required attributes for the input stream of
 // the WebSocketSink operator.
-type WebSocketSinkSendData_t = rstring strData, blob blobData, list<rstring> sendToUrlContextPaths; 
+type WebSocketSinkSendData_t = rstring strData, blob blobData, 
+	list<rstring> sendToUrlContextPaths, int32 parallelChannel; 
 
 public composite STTGatewayUtils {
 	graph

samples/STTGatewayUtils/com.ibm.streamsx.sttgateway.utils/STTGatewayUtils.spl

@@ -4,7 +4,7 @@
  <info:identity>
    <info:name>STTGatewayUtils</info:name>
    <info:description>A utility composite used by the VgwDataRouter and the speech processor applications.</info:description>
-   <info:version>1.0.3</info:version>
+   <info:version>1.0.4</info:version>
    <info:requiredProductVersion>4.2.1.6</info:requiredProductVersion>
  </info:identity>
  <info:dependencies/>

samples/STTGatewayUtils/info.xml

@@ -8,7 +8,7 @@
 /*
 ==============================================
 First created on: Nov/24/2020
-Last modified on: Sep/15/2021
+Last modified on: May/10/2022
 
 A) What does this example application do?
    --------------------------------------
@@ -573,7 +573,17 @@ public composite VgwDataRouter {
         			// whether it is sending binary speech data or an EOCS.
         			if(BSD.endOfCallSignal == false) {
         				// The incoming tuple contains binary speech data.
-        				//
+        				// If we are sent an empty speech blob, let us ignore that as
+        				// it will cause unwanted side effects in the downstream logic.
+        				if(blobSize(BSD.speech) <= 0ul) {
+							appTrc(Trace.error, 
+								"_ZZZZZ Empty speech packet received for " +
+								"vgwSessionId_vgwVoiceChannelNumber: " + _key + 
+								" We are going to ignore this packet by not sending it to" +
+								" a speech processor.");
+        					return;
+        				}
+        				
 						// We have to first check if this speech data belongs to a 
 						// brand new voice call or an already ongoing voice call.
 						if(has(_vgwSessionIdToSpeechProcessorMap, BSD.vgwSessionId) == false) {

samples/VgwDataRouter/com.ibm.streamsx.sttgateway.sample/VgwDataRouter.spl

@@ -4,21 +4,21 @@
  <info:identity>
    <info:name>VgwDataRouter</info:name>
    <info:description>Example that shows how to route VGW speech data to different Speech processor jobs</info:description>
-   <info:version>1.0.4</info:version>
+   <info:version>1.0.5</info:version>
    <info:requiredProductVersion>4.2.1.6</info:requiredProductVersion>
  </info:identity>
  <info:dependencies>
    <info:toolkit>
      <common:name>com.ibm.streamsx.sttgateway</common:name>
-     <common:version>[2.3.1,7.0.0]</common:version>
+     <common:version>[2.3.3,7.0.0]</common:version>
    </info:toolkit>
    <info:toolkit>
      <common:name>com.ibm.streamsx.websocket</common:name>
      <common:version>[1.0.9,7.0.0]</common:version>
    </info:toolkit>
    <info:toolkit>
      <common:name>STTGatewayUtils</common:name>
-     <common:version>[1.0.1,7.0.0]</common:version>
+     <common:version>[1.0.4,7.0.0]</common:version>
    </info:toolkit>
  </info:dependencies>
 </info:toolkitInfoModel>