Merge pull request #2371 from AllenInstitute/feature/2371-update-zeromq-xop

Update ZeroMQ XOP and related house keeping

Author: t-b

Packages/MIES/MIES_IVSCC.ipf

@@ -13,7 +13,7 @@
 /// - Listening port for the REP/ROUTER socket starts at #ZEROMQ_BIND_REP_PORT.
 /// - Listening port for the PUBLISHER socket starts at #ZEROMQ_BIND_PUB_PORT
 /// - If one of those ports is already in use, the next larger port is tried.
-/// - The publisher socket does include an automatic heartbeat message every 5 seconds. Subscribe to #ZeroMQ_HEARTBEAT if
+/// - The publisher socket does include an automatic heartbeat message every 5 seconds. Subscribe to #ZMQ_HEARTBEAT if
 ///   you want to receive that.
 /// - All available message filters can be queried via FFI_GetAvailableMessageFilters().
 /// - More information regarding the ZeroMQ-XOP is located [here](https://github.com/AllenInstitute/ZeroMQ-XOP/#readme)

Packages/MIES/MIES_MiesUtilities_Conversion.ipf

@@ -263,3 +263,15 @@ Function MapAnaFuncToConstant(string anaFunc)
 #endif // AUTOMATED_TESTING
 	endswitch
 End
+
+/// @brief returns the unit string for the AD channel depending on clampmode
+threadsafe Function/S GetADChannelUnit(variable clampMode)
+
+	return SelectString(clampMode == V_CLAMP_MODE, "mV", "pA")
+End
+
+/// @brief returns the unit string for the DA channel depending on clampmode
+threadsafe Function/S GetDAChannelUnit(variable clampMode)
+
+	return SelectString(clampMode == V_CLAMP_MODE, "pA", "mV")
+End

Packages/MIES/MIES_MiesUtilities_ZeroMQ.ipf

@@ -11,7 +11,7 @@
 
 Function GetZeroMQXOPFlags()
 
-	return ZeroMQ_SET_FLAGS_DEFAULT | ZeroMQ_SET_FLAGS_LOGGING | ZeroMQ_SET_FLAGS_NOBUSYWAITRECV
+	return ZeroMQ_SET_FLAGS_DEFAULT | ZeroMQ_SET_FLAGS_NOBUSYWAITRECV
 End
 
 /// @brief Start the ZeroMQ sockets and the message handler

Packages/MIES/MIES_Oscilloscope.ipf

@@ -581,7 +581,7 @@ Function SCOPE_UpdateOscilloscopeData(string device, variable dataAcqOrTP, [vari
 					endif
 					tpInput.clampAmp  = clampAmp
 					tpInput.clampMode = hsProp[headstage][%ClampMode]
-					tpInput.hsIndex   = headstage
+					tpInput.headstage = headstage
 
 					DEBUGPRINT("headstage: ", var = headstage)
 					DEBUGPRINT("channel: ", var = numDACs + j)

Packages/MIES/MIES_Publish.ipf

@@ -25,20 +25,26 @@ static Function PUB_GetJSONTemplate(string device, variable headstage)
 End
 
 /// @brief Publish the given message as given by the JSON and the filter
-static Function PUB_Publish(variable jsonID, string messageFilter)
+threadsafe Function PUB_Publish(variable jsonID, string messageFilter, [variable releaseJSON])
 
 	variable err
-	string   payload
 
-	payload = JSON_Dump(jsonID)
-	JSON_Release(jsonID)
+	releaseJSON = ParamIsDefault(releaseJSON) ? 1 : !!releaseJSON
+
+	Make/T/FREE filter = {messageFilter}
+	Make/T/FREE payload = {JSON_Dump(jsonID)}
+	Make/FREE/WAVE wv = {filter, payload}
+
+	if(releaseJSON)
+		JSON_Release(jsonID)
+	endif
 
 	AssertOnAndClearRTError()
 	try
-		zeromq_pub_send(messageFilter, payload); AbortOnRTE
+		zeromq_pub_send_multi(wv); AbortOnRTE
 	catch
 		err = ClearRTError()
-		BUG("Could not publish " + messageFilter + " due to: " + num2str(err))
+		BUG_TS("Could not publish " + messageFilter + " due to: " + num2str(err))
 	endtry
 End
 

Packages/MIES/MIES_Structures.ipf

@@ -303,7 +303,7 @@ Structure TPAnalysisInput
 	variable baselineFrac
 	variable tpLengthPoints
 	variable readTimeStamp
-	variable hsIndex
+	variable headstage
 	string device
 	variable measurementMarker
 	variable activeADCs

Packages/MIES/MIES_TestPulse.ipf

@@ -192,7 +192,7 @@ Function TP_ROAnalysis(STRUCT ASYNC_ReadOutStruct &ar)
 
 	WAVE/SDFR=dfr inData     = outData
 	NVAR/SDFR=dfr now        = now
-	NVAR/SDFR=dfr hsIndex    = hsIndex
+	NVAR/SDFR=dfr headstage  = headstage
 	SVAR/SDFR=dfr device     = device
 	NVAR/SDFR=dfr marker     = marker
 	NVAR/SDFR=dfr activeADCs = activeADCs
@@ -218,11 +218,11 @@ Function TP_ROAnalysis(STRUCT ASYNC_ReadOutStruct &ar)
 		asyncBuffer[bufSize][posAsync][posMarker]     = marker
 	endif
 
-	asyncBuffer[i][posBaseline][hsIndex] = inData[%BASELINE]
-	asyncBuffer[i][posSSRes][hsIndex]    = inData[%STEADYSTATERES]
-	asyncBuffer[i][posInstRes][hsIndex]  = inData[%INSTANTRES]
-	asyncBuffer[i][posElevSS][hsIndex]   = inData[%ELEVATED_SS]
-	asyncBuffer[i][posElevInst][hsIndex] = inData[%ELEVATED_INST]
+	asyncBuffer[i][posBaseline][headstage] = inData[%BASELINE]
+	asyncBuffer[i][posSSRes][headstage]    = inData[%STEADYSTATERES]
+	asyncBuffer[i][posInstRes][headstage]  = inData[%INSTANTRES]
+	asyncBuffer[i][posElevSS][headstage]   = inData[%ELEVATED_SS]
+	asyncBuffer[i][posElevInst][headstage] = inData[%ELEVATED_INST]
 
 	asyncBuffer[i][posAsync][%NOW]           = now
 	asyncBuffer[i][posAsync][%REC_CHANNELS] += 1
@@ -894,7 +894,7 @@ threadsafe Function/DF TP_TSAnalysis(DFREF dfrInp)
 	NVAR/SDFR=dfrInp baselineFrac     = param4
 	NVAR/SDFR=dfrInp lengthTPInPoints = param5
 	NVAR/SDFR=dfrInp now              = param6
-	NVAR/SDFR=dfrInp hsIndex          = param7
+	NVAR/SDFR=dfrInp headstage        = param7
 	SVAR/SDFR=dfrInp device           = param8
 	NVAR/SDFR=dfrInp marker           = param9
 	NVAR/SDFR=dfrInp activeADCs       = param10
@@ -994,7 +994,7 @@ threadsafe Function/DF TP_TSAnalysis(DFREF dfrInp)
 
 	// additional data copy
 	variable/G dfrOut:now        = now
-	variable/G dfrOut:hsIndex    = hsIndex
+	variable/G dfrOut:headstage  = headstage
 	string/G   dfrOut:device     = device
 	variable/G dfrOut:marker     = marker
 	variable/G dfrOut:activeADCs = activeADCs
@@ -1505,7 +1505,7 @@ Function/DF TP_PrepareAnalysisDF(string device, STRUCT TPAnalysisInput &tpInput)
 	ASYNC_AddParam(threadDF, var = tpInput.baselineFrac)
 	ASYNC_AddParam(threadDF, var = tpInput.tpLengthPoints)
 	ASYNC_AddParam(threadDF, var = tpInput.readTimeStamp)
-	ASYNC_AddParam(threadDF, var = tpInput.hsIndex)
+	ASYNC_AddParam(threadDF, var = tpInput.headstage)
 	ASYNC_AddParam(threadDF, str = tpInput.device)
 	ASYNC_AddParam(threadDF, var = tpInput.measurementMarker)
 	ASYNC_AddParam(threadDF, var = tpInput.activeADCs)

Packages/MIES/MIES_WaveDataFolderGetters.ipf

@@ -162,10 +162,10 @@ Function/WAVE GetChanAmpAssignUnit(string device)
 		Make/T/N=(4, NUM_HEADSTAGES) dfr:ChanAmpAssignUnit/WAVE=wv
 		wv = ""
 
-		wv[0][] = "mV"
-		wv[1][] = "pA"
-		wv[2][] = "pA"
-		wv[3][] = "mV"
+		wv[0][] = GetDAChannelUnit(V_CLAMP_MODE)
+		wv[1][] = GetADChannelUnit(V_CLAMP_MODE)
+		wv[2][] = GetDAChannelUnit(I_CLAMP_MODE)
+		wv[3][] = GetADChannelUnit(I_CLAMP_MODE)
 	endif
 
 	SetDimLabel ROWS, 0, VC_DAUnit, wv
@@ -494,7 +494,7 @@ End
 /// Column 8: average elevated level (instantaneous)
 ///
 /// Layers:
-/// - NUM_HEADSTAGES positions with value entries at hsIndex
+/// - NUM_HEADSTAGES
 Function/WAVE GetTPResultAsyncBuffer(string device)
 
 	variable versionOfNewWave = 1

Packages/MIES/ZeroMQ_Interop.ipf

@@ -1,9 +1,11 @@
-#pragma TextEncoding = "UTF-8"
-#pragma rtGlobals=3		// Use modern global access method and strict wave access.
-#pragma IgorVersion=8.0
+#pragma TextEncoding="UTF-8"
+#pragma rtGlobals=3 // Use modern global access method and strict wave access.
+#pragma IgorVersion=6.37
 
 // This file is part of the `ZeroMQ-XOP` project and licensed under BSD-3-Clause.
 
+/// ** Define ZeroMQ Constants (for back-compatibility) for Igor8 case. **
+#if igorVersion() >= 8
 /// @name Flags for zeromq_set()
 /// @anchor ZeroMQSetFlags
 ///@{
@@ -36,25 +38,62 @@ Constant ZeroMQ_INVALID_MESSAGE_FORMAT    = 10009
 Constant ZeroMQ_INVALID_LOGGING_TEMPLATE  = 10010
 Constant ZeroMQ_MESSAGE_FILTER_DUPLICATED = 10011
 Constant ZeroMQ_MESSAGE_FILTER_MISSING    = 10012
+Constant ZeroMQ_MESSAGE_INVALID_TYPE      = 10013
 ///@}
+#endif
 
-Constant REQ_SUCCESS                         =   0
-Constant REQ_UNKNOWN_ERROR                   =   1
-Constant REQ_INVALID_JSON_OBJECT             =   3
-Constant REQ_INVALID_VERSION                 =   4
-Constant REQ_INVALID_OPERATION               =   5
-Constant REQ_INVALID_OPERATION_FORMAT        =   6
-Constant REQ_INVALID_MESSAGEID               =   7
-Constant REQ_OUT_OF_MEMORY                   =   8
+/// @name Flags for zeromq_set()
+/// @anchor ZeroMQSetFlags
+///@{
+/// Sets the default flags (no debug, no ipv6, busy wait on receive)
+Constant ZMQ_SET_FLAGS_DEFAULT = 0x1
+/// Enable debug output
+Constant ZMQ_SET_FLAGS_DEBUG = 0x2
+/// Enable ipv6 support
+Constant ZMQ_SET_FLAGS_IPV6 = 0x4
+/// Don't do busy waiting on zeromq_server_recv() and zeromq_client_recv()
+/// instead immediately return if no messages are available.
+Constant ZMQ_SET_FLAGS_NOBUSYWAITRECV = 0x8
+/// Log incoming and outgoing messages
+Constant ZMQ_SET_FLAGS_LOGGING = 0x10
+
+///@}
+
+StrConstant ZMQ_HEARTBEAT = "heartbeat"
+
+/// @name Error codes
+/// @anchor ZeroMQErrorCodes
+///@{
+Constant ZMQ_UNKNOWN_SET_FLAG          = 10003
+Constant ZMQ_INTERNAL_ERROR            = 10004
+Constant ZMQ_INVALID_ARG               = 10005
+Constant ZMQ_HANDLER_ALREADY_RUNNING   = 10006
+Constant ZMQ_HANDLER_NO_CONNECTION     = 10007
+Constant ZMQ_MISSING_PROCEDURE_FILES   = 10008
+Constant ZMQ_INVALID_MESSAGE_FORMAT    = 10009
+Constant ZMQ_INVALID_LOGGING_TEMPLATE  = 10010
+Constant ZMQ_MESSAGE_FILTER_DUPLICATED = 10011
+Constant ZMQ_MESSAGE_FILTER_MISSING    = 10012
+Constant ZMQ_MESSAGE_INVALID_TYPE      = 10013
+///@}
+
+Constant REQ_SUCCESS                  = 0
+Constant REQ_UNKNOWN_ERROR            = 1
+Constant REQ_INVALID_JSON_OBJECT      = 3
+Constant REQ_INVALID_VERSION          = 4
+Constant REQ_INVALID_OPERATION        = 5
+Constant REQ_INVALID_OPERATION_FORMAT = 6
+Constant REQ_INVALID_MESSAGEID        = 7
+Constant REQ_OUT_OF_MEMORY            = 8
 // error codes for CallFunction class
-Constant REQ_PROC_NOT_COMPILED               = 100
-Constant REQ_NON_EXISTING_FUNCTION           = 101
-Constant REQ_TOO_FEW_FUNCTION_PARAMS         = 102
-Constant REQ_TOO_MANY_FUNCTION_PARAMS        = 103
-Constant REQ_UNSUPPORTED_FUNC_SIG            = 104
-Constant REQ_UNSUPPORTED_FUNC_RET            = 105
-Constant REQ_INVALID_PARAM_FORMAT            = 106
-Constant REQ_FUNCTION_ABORTED                = 107
+Constant REQ_PROC_NOT_COMPILED        = 100
+Constant REQ_NON_EXISTING_FUNCTION    = 101
+Constant REQ_TOO_FEW_FUNCTION_PARAMS  = 102
+Constant REQ_TOO_MANY_FUNCTION_PARAMS = 103
+Constant REQ_UNSUPPORTED_FUNC_SIG     = 104
+Constant REQ_UNSUPPORTED_FUNC_RET     = 105
+Constant REQ_INVALID_PARAM_FORMAT     = 106
+Constant REQ_FUNCTION_ABORTED         = 107
 
 /// @name Functions which might be useful for outside callers
 /// @anchor ZeroMQInterfaceFunctions

Packages/tests/Basic/UTF_ZeroMQPublishing.ipf

@@ -3,49 +3,13 @@
 #pragma rtFunctionErrors=1
 #pragma ModuleName=ZeroMQPublishingTests
 
-// #define OUTPUT_DOCUMENTATION_JSON_DUMP
-
 static Function TEST_CASE_BEGIN_OVERRIDE(string testname)
 
 	TestCaseBeginCommon(testname)
 
 	PrepareForPublishTest()
 End
 
-static Function FetchAndParseMessage(string filter)
-
-	variable jsonID
-	string   msg
-
-	msg = FetchPublishedMessage(filter)
-
-	CHECK_PROPER_STR(msg)
-
-	jsonID = JSON_Parse(msg)
-	CHECK_GE_VAR(jsonID, 0)
-
-#ifdef OUTPUT_DOCUMENTATION_JSON_DUMP
-	WAVE/T contents = ListToTextWave(JSON_Dump(jsonID, indent = 2), "\n")
-
-	contents[] = "///    " + contents[p]
-
-	print "/// Filter: #XXXX"
-	print "///"
-	print "/// Example:"
-	print "///"
-	print "/// \\rst"
-	print "/// .. code-block:: json"
-	print "///"
-	for(s : contents)
-		print s
-	endfor
-	print "///"
-	print "/// \\endrst"
-#endif // OUTPUT_DOCUMENTATION_JSON_DUMP
-
-	return jsonID
-End
-
 static Function CheckPressureState()
 
 	string device, expected, actual
@@ -413,7 +377,7 @@ End
 
 static Function CheckAccessResSmoke()
 
-	string device, msg, expected, actual
+	string device, expected, actual
 	variable headstage, i, jsonID, value, sweepNo
 
 	device    = "my_device"
@@ -439,9 +403,7 @@ static Function CheckAccessResSmoke()
 
 	MIES_PUB#PUB_AccessResistanceSmoke(device, sweepNo, headstage)
 
-	msg = FetchPublishedMessage(ANALYSIS_FUNCTION_AR)
-
-	jsonID = JSON_Parse(msg)
+	jsonID = FetchAndParseMessage(ANALYSIS_FUNCTION_AR)
 
 	expected = LABNOTEBOOK_BINARY_UNIT
 	actual   = JSON_GetString(jsonID, "/results/USER_Access Res. Smoke Set QC/unit")

Packages/tests/UTF_HardwareHelperFunctions.ipf

@@ -837,7 +837,7 @@ End
 
 Function CheckPublishedMessage(string device, variable type)
 
-	string expectedFilter, msg
+	string   expectedFilter
 	variable jsonID
 
 	switch(type)
@@ -858,9 +858,7 @@ Function CheckPublishedMessage(string device, variable type)
 			return NaN
 	endswitch
 
-	msg = FetchPublishedMessage(expectedFilter)
-	CHECK_PROPER_STR(msg)
-	jsonID = JSON_Parse(msg)
+	jsonID = FetchAndParseMessage(expectedFilter)
 	CHECK_GE_VAR(jsonID, 0)
 	JSON_Release(jsonID)
 End
@@ -1666,12 +1664,10 @@ End
 
 Function CheckStartStopMessages(string mode, string state)
 
-	string msg, actual, expected
+	string actual, expected
 	variable jsonID
 
-	msg = FetchPublishedMessage(DAQ_TP_STATE_CHANGE_FILTER)
-
-	jsonID   = JSON_Parse(msg)
+	jsonID   = FetchAndParseMessage(DAQ_TP_STATE_CHANGE_FILTER)
 	actual   = JSON_GetString(jsonID, "/" + mode)
 	expected = state
 	CHECK_EQUAL_STR(actual, expected)