Publishing and Subscribing with the Simulation API
Introduction to the Simulation API
The Simulation API is used for all actions related to a power system simulation. It is used to start, pause, restart, and stop a simulation from the command line or inside an application. It is all used to subscribe to measurements of equipment (such as line flows, loads, and DG setpoints) and the statuses of switches, capacitors, transformer taps, etc. It is also used to publish equipment control and other simulation input commands.
In the Application Components diagram (explained in detail with sample code in GridAPPS-D Application Structure), the PowerGrid Models API is used for controlling the simulation, subscribing to measurement data, and controlling equipment.
This section covers only the portion of the API used for subscribing to measurements and publishing equipment control commands. Usage of the API for starting, stopping, and pausing simulations is covered in Creating and Running Simulations with Simulation API
Processing Measurements & App Core Algorithm
The central portion of a GridAPPS-D application is the measurement processing and core algorithm section. This section is built as either a class or function definition with prescribed arguments. Each has its advantages and disadvantages:
The function-based approach is simpler and easier to implement. However, any parameters obtained from other APIs or methods to be used inside the function currently need to be defined as global variables.
The class-based approach is more complex, but also more powerful. It provides greater flexibility in creating additional methods, arguments, etc.
App Core Information Flow
This portion of the application does not communicate directly with the GridAPPS-D platform.
Instead, the next part of the GridAPPS-D application (Subscribing to Simulation Output) delivers the simulated SCADA measurement data to the core algorithm function / class definition. The core algorithm processes the data to extract the desired measurements and run its optimization / control agorithm.
Structure of Simulation Output Message
The first part of the application core is parsing simulated SCADA and measurement data that is delivered to the application.
The general format of the messages received by the Simulation API is a python dictionary with the following key-value pairs:
{
"simulation_id" : string,
"message" : {
"timestamp" : epoch time number,
"measurements" : {
"meas mrid 1":{
"PNV measurement_mrid": "meas mrid 1"
"magnitude": number,
"angle": number },
"meas mrid 2":{
"VA measurement_mrid": "meas mrid 2"
"magnitude": number,
"angle": number },
"meas mrid 3":{
"Pos measurement_mrid": "meas mrid 3"
"value": number },
.
.
.
"meas mrid n":{
"measurement_mrid": "meas mrid n"
"magnitude": number,
"angle": number },
}
}
Format of Measurement Values
In the message above, note the difference in the key-value pair structure of different types of measurements:
PNV Voltage Measurements
These are specified as magnitude and angle key-value pairs.
Magnitude is the RMS phase-to-neutral voltage.
Angle is phase angle of the voltage at the particular node.
VA Volt-Ampere Apparent Power Measurements
These are specified as magnitude and angle key-value pairs. * Magnitude is the apparent power. * Angle is complex power triangle angle (i.e. acos(power factor))
Pos Position Measurements
These are specified as a value key-value pair. * Value is the position of the particular measurement * For switch objects: value = 1 means “closed”, value = 0 means “open” * For capacitor objects, values are reversed: value = 1 means “on”, value = 0 means “off” * For regulator objects, value is the tap position, ranging from -16 to 16
Role of Measurement mRIDs
The simulation output message shown above typically contains the measurement mRIDs for all available sensors for all equipment in the power system model. The application needs to filter the simulation output message to just the set of measurements relevant to the particular application (e.g. switch positions for a FLISR app or regulator taps for a VVO app).
The equipment and measurement mRIDs are obtained in the first two sections of the application. See Query for Power System Model and Query for Measurement mRIDs for examples of how these code sections fit in a sample app.
API syntax details for the query messages to PowerGrid Models API to obtain equipment info and measurement mRIDs are given in Query for Object Dictionary and Query for Measurements.
These mRIDs will be needed to parse the simulation output message and filter it to just the desired set of measurements.
For the example below, we will be interested in only the measurement associated with switches, so we will use the PowerGrid Models API to query for the set of measurements associated with the CIM Class LoadBreakSwitch. We then will filter those values to just the mRIDs associated with each type of measurement.
[ ]:
from gridappsd import topics as t
# Create query message to obtain measurement mRIDs for all switches
message = {
"modelId": model_mrid,
"requestType": "QUERY_OBJECT_MEASUREMENTS",
"resultFormat": "JSON",
"objectType": "LoadBreakSwitch"
}
# Pass query message to PowerGrid Models API
response_obj = gapps.get_response(t.REQUEST_POWERGRID_DATA, message)
measurements_obj = response_obj["data"]
# Switch position measurements (Pos)
Pos_obj = [k for k in measurements_obj if k['type'] == 'Pos']
# Switch phase-neutral-voltage measurements (PNV)
PNV_obj = [k for k in measurements_obj if k['type'] == 'PNV']
# Switch volt-ampere apparent power measurements (VA)
VA_obj = [k for k in measurements_obj if k['type'] == 'VA']
# Switch current measurements (A)
A_obj = [k for k in measurements_obj if k['type'] == 'A']
App Core as a Function Definition
The first approach used to build the application core is to define a function with the correct set of arguments that is then passed to the .subscribe() method associated with the GridAPPPSD() object.
The function does not require a specific name, and is somewhat easier to define and use. However, the arguments of the function need to be named correctly for the GridAPPSD-Python library to process the simulation output correctly.
The format for the function definition is
def mySubscribeFunction(header, message):
# do something when receive a message
# parse to get measurments
# do some calculations
# publish some equipment commands
# display some results
That function handle is then passed as an argument to the .subscribe(topic, function_handle) method when subscribing to the simulation in the next section.
Note that the subscription function definition does not allow any additional parameters to be passed. The only allowed arguments are header and message.
Any other parameters, such as measurement mRIDs will need to be defined as global variables.
[ ]:
# Define global python dictionary of position measurements
global Pos_obj
Pos_obj = [k for k in measurements_obj if k['type'] == 'Pos']
# Define global python dictionary of phase-neutral-voltage measurements (PNV)
global PNV_obj
PNV_obj = [k for k in measurements_obj if k['type'] == 'PNV']
# Define global python dictionary of volt-ampere apparent power measurements (VA)
VA_obj = [k for k in measurements_obj if k['type'] == 'VA']
# Current measurements (A)
A_obj = [k for k in measurements_obj if k['type'] == 'A']
Below is the sample code for the core section of a basic application that tracks the number of open switches and the number of switches that are outaged.
[ ]:
# Only allowed arguments are `header` and `message`
# message is simulation output message in format above
def DemoAppCoreFunction(header, message):
# Extract time and measurement values from message
timestamp = message["message"]["timestamp"]
meas_value = message["message"]["measurements"]
# Obtain list of all mRIDs from message
meas_mrid = list(meas_value.keys())
# Example 1: Count the number of open switches
open_switches = []
for index in Pos_obj:
if index["measid"] in meas_value:
mrid = index["measid"]
power = meas_value[mrid]
if power["value"] == 0: # Filter to measurements with value of zero
open_switches.append(index["eqname"])
# Print message to command line
print("............")
print("Number of open switches at time", timestamp, ' is ', len(set(open_switches)))
# Example 2: Count the number of outaged switches (voltage = 0)
dead_switches = []
for index in PNV_obj:
if index["measid"] in meas_value:
mrid = index["measid"]
voltage = meas_value[mrid]
if voltage["magnitude"] == 0.0:
dead_switches.append(index["eqname"])
# Print message to command line
print("............")
print("Number of outaged switches at time", timestamp, ' is ', len(set(dead_switches)))
App Core as a Class Definition
The second approach used to build the app core and process measurements is to define a class containing two methods named __init__ and on_message.
These methods specify 1) how your app would initialize variables and attributes at the start of the simulation and 2) how your app behaves when it receives various messages.
IMPORTANT! The GridAPPS-D Platform uses the exact names and syntax for the methods:
__init__(self, simulation_id, gapps_object, optional_objects)– This method requires the simulation_id and GridAPPS-D connection object. It is also possible add other user-defined arguments, such as measurement mRIDs or other information required by your application.on_message(self, headers, message)– This method allows the class to subscribe to simulation measurements. It also contains the core behavior of your application and how it responds to each type of message.
It is also possible to use the same class definition to subscribe to other topics, such as Simulation Logs. This is done by creating additional user-defined methods and then passing those methods to the .subcribe() method associated with the GridAPPS-D connection object. An example of how this is done is provided for subcribing to simulation logs in Logging with a Class Method.
class YourSimulationClassName(object):
# Your documentation text here on what app does
def __init__(self, simulation_id, gapps_obj, meas_obj, your_obj):
# Instantiate class with specific initial state
# Attributes required by Simulation API
self._gapps = gapps_obj
self._simulation_id = simulation_id
# Attributes to publish difference measurements
self.diff = DifferenceBuilder(simulation_id)
# Custom attributes for measurements, custom info
self.meas_mrid = meas_obj
self.your_attribute1 = your_obj["key1"]
self.your_attribute2 = your_obj["key2"]
def on_message(self, headers, message):
# What app should do when it receives a subscription message
variable1 = message["message"]["key1"]
variable2 = message["message"]["key2"]
# Insert your custom app behavior here
if variable1 == foo:
bar = my_optimization_result
# Insert your custom equipment commands here
if variable2 == bar:
self.diff.add_difference(object_mrid, control_attribute, new_value, old_value)
def my_custom_method_1(self, headers, message):
# Use extra methods to subscribe to other topics, such as simulation logs
variable1 = message["key1"]
variable2 = message["key2"]
def my_custom_method_2(self, param1, param2):
# Use extra methods as desired
variable1 = foo
variable2 = bar
Below is the sample code for the core section of a basic application that tracks the number of open switches and the number of switches that are outaged.
[ ]:
# Application core built as a class definition
class DemoAppCoreClass(object):
# Subscription callback from GridAPPSD object
def __init__(self, simulation_id, gapps_obj, meas_obj):
self._gapps = gapps_obj # GridAPPS-D connection object
self._simulation_id = simulation_id # Simulation ID
self.meas_mrid = meas_obj # Dictionary of measurement mRIDs obtained earlier
def on_message(self, headers, message):
# Extract time and measurement values from message
timestamp = message["message"]["timestamp"]
meas_value = message["message"]["measurements"]
# Filter measurement mRIDs for position and voltage sensors
Pos_obj = [k for k in self.meas_mrid if k['type'] == 'Pos']
PNV_obj = [k for k in self.meas_mrid if k['type'] == 'PNV']
# Example 1: Count the number of open switches
open_switches = []
for index in Pos_obj:
if index["measid"] in meas_value:
mrid = index["measid"]
power = meas_value[mrid]
if power["value"] == 0: # Filter to measurements with value of zero
open_switches.append(index["eqname"])
# Print message to command line
print("............")
print("Number of open switches at time", timestamp, ' is ', len(set(open_switches)))
# Example 2: Count the number of outaged switches (voltage = 0)
dead_switches = []
for index in PNV_obj:
if index["measid"] in meas_value:
mrid = index["measid"]
voltage = meas_value[mrid]
if voltage["magnitude"] == 0.0:
dead_switches.append(index["eqname"])
# Print message to command line
print("............")
print("Number of outaged switches at time", timestamp, ' is ', len(set(dead_switches)))
Subscribing to Simulation Output
Simulation Subscription Information Flow
The figure below shows the information flow involved in subscribing to the simulation output.
The subscription request is sent using gapps.subscribe(topic, class/function object) on the specific Simulation topic channel (explained in API Communication Channels). No immediate response is expected back from the platform. However, after the next simulation timestep, the Platform will continue to deliver a complete set of measurements back to the application for each timestep until the end of the simulation.
Application passes subscription request to GridAPPS-D Platform
The subscription request is perfromed by passing the app core algorithm function / class definition to the gapps.subscribe method. The application then passes the subscription request through the Simulation API to the topic channel for the particular simulation on the GOSS Message Bus. If the application is authorized to access simulation output, the subscription request is delivered to the Simulation Manager.
GridAPPS-D Platform delivers published simulation output to Application
Unlike the previous queries made to the various databases, the GridAPPS-D Platform does not provide any immediate response back to the application. Instead, the Simulation Manager will start delivering measurement data back to the application through the Simulation API at each subsequent timestep until the simulation ends or the application unsubscribes. The measurement data is then passed to the core algorithm class / function, where it is processed and used to run the app’s optimization / control algorithms.
Subscription API Communication Channel
This is a dynamic /topic/ communication channel that is best implemented by importing the GriAPPSD-Python library function for generating the correct topic. This communication channel is used for all simulation subscription API calls.
[ ]:
from gridappsd.topics import simulation_output_topic
output_topic = simulation_output_topic(viz_simulation_id)
Comparison of Subscription Approaches
Each approach has its advantages and disadvantages.
The function-based approach is simpler and easier to implement. However, any parameters obtained from other APIs or methods to be used inside the function currently need to be defined as global variables.
The class-based approach is more complex, but also more powerful. It provides greater flexibility in creating additional methods, arguments, etc.
The Simulation Library-based approach is easiest, but only works currently for parallel digital twin simulations started using the
simulation_obj.start_simulation()method.
The choice of which approach is used depends on the personal preferences of the application developer.
Subscription for Function-based App Core
If the application core was created as a function definition as shown in App Core as Function Definition, then the function name is passed to the .subscribe(output_topic, core_function) method of the GridAPPS-D Connection object.
[ ]:
conn_id = gapps.subscribe(output_topic, DemoAppCoreFunction)
Subscription for Class-Based App Core
If the application core was created as a class definition as shown in App Core as Class Definition, then the function name is passed to the .subscribe(output_topic, object) method of the GridAPPS-D connection object.
After defining the class for the application core as shown above, we create another object that will be passed to the subscription method. The required parameters for this object are the same as those defined for the __init__() method of the app core class, typically the Simulation ID, GridAPPS-D connection object, dictionary of measurements needed by the app core, and any user-defined objects.
class_obj = AppCoreClass(simulation_id, gapps_obj, meas_obj, your_obj)
[ ]:
demo_obj = DemoAppCoreClass(viz_simulation_id, gapps, measurements_obj)
conn_id = gapps.subscribe(subscribe_topic, demo_obj)
If we wish to subscribe to an additional topic (such as the Simulation Logs, a side communication channel between two different applications, or a communication with a particular service), we can define an additional method in the class (such as my_custom_method_1 in the example class definition above) and then pass it to to the .subscribe(topic, object.method) method associated with the GridAPPS-D connection object:
gapps.subscribe(other_topic, demo_obj.my_custom_method_1)
Subscribing to Parallel Simulations
Parallel simulations started using the Simulation API (as shown in Starting a Simulation) and the Simulation library in GridAPPSD-Python do not need to use the gapps.subscribe method.
Instead, the GridAPPSD-Python library contains several shortcut functions that can be used. These methods currently cannot interact with a simulation started from the Viz. This functionality will be added in a future release.
The code block below shows how a parallel simulation can be started using a simulation start message stored in a JSON file. The simulation is started using the .start_simulation() method.
[ ]:
import json, os
from gridappsd import GridAPPSD
from gridappsd.simulation import Simulation
# Connect to GridAPPS-D Platform
os.environ['GRIDAPPSD_USER'] = 'tutorial_user'
os.environ['GRIDAPPSD_PASSWORD'] = '12345!'
gapps = GridAPPSD()
assert gapps.connected
model_mrid = "_C1C3E687-6FFD-C753-582B-632A27E28507"
run123_config = json.load(open("Run123NodeFileSimAPI.json")) # Pull simulation config from saved file
simulation_obj = Simulation(gapps, run123_config) # Create Simulation object
simulation_obj.start_simulation() # Start Simulation
print("Successfully started simulation with simulation_id: ", simulation_obj.simulation_id)
[ ]:
simulation_id = simulation_obj.simulation_id
The Simulation library provides four methods that can be used to define how the platform interacts with the simulation:
.add_ontimestep_callback(myfunction1)– Run the desired function on each timestep.add_onmesurement_callback(myfunction2)– Run the desired function when a measurement is received..add_oncomplete_callback(myfunction3)– Run the desired function when simulation is finished.add_onstart_callback(myfunction4)– Run desired function when simulation is started
Note: method name ``.add_onmesurement_callback`` is misspelled in the library definition!!
Note that the measurement callback method returns just the measurements and timestamps without any of the message formatting used in the messages received by using the gapps.subscribe(output_topic, object) approach.
The python dictionary returned by the GridAPPS-D Simulation output to the .add_onmesurement_callback() method is always named measurements and uses the following key-value pairs format:
{
'_pnv_meas_mrid_1': {'angle': number,
'magnitude': number,
'measurement_mrid': '_pnv_meas_mrid_1'},
'_va_meas_mrid_2': { 'angle': number,
'magnitude': number,
'measurement_mrid': '_va_meas_mrid_2'},
'_pos_meas_mrid_3': {'measurement_mrid': '_pos_meas_mrid_3',
'value': 1},
.
.
.
'_pnv_meas_mrid_n': {'angle': number,
'magnitude': number,
'measurement_mrid': '_pnv_meas_mrid_1'}
}
To use use these methods, we define a set of functions that determine the behavior of the application for each of the four types of callbacks listed above. These functions are similar to those defined for the function-based app core algorithm.
def my_onstart_func(sim):
# Do something when the simulation starts
# Do something else when the sim starts
simulation_obj.add_onstart_callback(my_onstart_func)
def my_onmeas_func(sim, timestamp, measurements):
# Do something when app receives a measurement
# Insert your custom app behavior here
if measurements[object_mrid] == foo:
bar = my_optimization_result
simulation_obj.add_onmesurement_callback(my_onmeas_func)
def my_oncomplete_func(sim):
# Do something when simulation is complete
# example: delete all variables, close files
simulation_obj.add_oncomplete_callback(my_oncomplete_func)
The code block below shows how the same app core algorithm can be used for a parallel simulation using the .add_onmesurement_callback() method:
[ ]:
def demo_onmeas_func(sim, timestamp, measurements):
open_switches = []
for index in Pos_obj:
if index["measid"] in measurements:
mrid = index["measid"]
power = measurements[mrid]
if power["value"] == 0:
open_switches.append(index["eqname"])
print("............")
print("Number of open switches at time", timestamp, ' is ', len(set(open_switches)))
[ ]:
simulation_obj.add_onmesurement_callback(demo_onmeas_func)
Publishing Commands to Simulation Input
The next portion of a GridAPPS-D App is publishing equipment control commands based on the optimization results or objectives of the app algorithm.
Depending on the preference of the developer, this portion can be a separate function definition, or included as part of the main class definition as part of the App Core as a Class Definition described earlier.
Equipment Command Information Flow
The figure below outlines information flow involved in publishing equipment commands to the simulation input.
Unlike the various queries to the databases in the app sections earlier, equipment control commands are passed to the GridAPPS-D API using the gapps.send(topic, message) method. No response is expected from the GridAPPS-D platform.
If the application desires to verify that the equipment control command was received and implemented, it needs to do so by 1) checking for changes in the associated measurements at the next timestep and/or 2) querying the Timeseries Database for historical simulation data associated with the equipment control command.
Application sends difference message to GridAPPS-D Platform
First, the application creates a difference message containing the current and desired future control point / state of the particular piece of power system equipment to be controlled. The difference message is a JSON string or equivalant Python dictionary object. The syntax of a difference message is explained in detail below in Format of Difference Message.
The application then passes the query through the Simulation API to the GridAPPS-D Platform, which publishes it on the topic channel for the particular simulation on the GOSS Message Bus. If the app is authenticated and authorized to control equipment, the difference message is delivered to the Simulation Manager. The Simulation Manager then passes the command to the simulation through the Co-Simulation Bridge (either FNCS or HELICS).
No response from GridAPPS-D Platform back to Application
The GridAPPS-D Platform does not provide any response back to the application after processing the difference message and implementing the new equipment control setpoint.
Simulation Input API Channel
This is a dynamic /topic/ communication channel that is best implemented by importing the GriAPPSD-Python library function for generating the correct topic.
from gridappsd.topics import simulation_input_topicinput_topic = simulation_input_topic(simulation_id)
[ ]:
from gridappsd.topics import simulation_input_topic
input_topic = simulation_input_topic(viz_simulation_id)
Equipment Control mRIDs
The mRIDs for controlling equipment are generally the same as those obtained using the QUERY_OBJECT_DICT key with the PowerGrid Models API, which was covered in Query for Object Dicionary.
However, the control attributes for each class of equipment in CIM use a different naming convention than those for the object types. Below is a list of "objectType" used to query for mRIDs and the associated control attribute used in a difference message for each category of power system equipment:
Switches
CIM Class Key:
"objectType": "LoadBreakSwitch"Control Attribute:
"attribute": "Switch.open"Values:
1is open,0is closed
Capacitor Banks:
CIM Class Key:
"objectType": "LinearShuntCompensator"Control Attribute:
"attribute": "ShuntCompensator.sections"Values:
0is off/open,1is on/closed
Inverter-based DERs:
CIM Class Key:
"objectType": "PowerElectronicsConnection"Control Attribute:
"attribute": "PowerElectronicsConnection.p"Control Attribute:
"attribute": "PowerElectronicsConnection.q"Values: number in Watts or VArs (not kW)
Synchronous Rotating (diesel) DGs:
CIM Class Key:
"objectType": "SynchronousMachine"Control Attribute:
"attribute": "RotatingMachine.p"Control Attribute:
"attribute": "RotatingMachine.q"Values: number in Watts or VArs (not kW)
Regulating Transformer Tap:
CIM Class Key:
"objectType": "RatioTapChanger"Control Attribute:
"attribute": "TapChanger.step"Values: integer value for tap step
The query for RatioTapChanger is not supported in the PowerGrid Models API at the current time. A custom SPARQL query needs to be done using the sample query in `CIMHub Sample Queries <https://github.com/GRIDAPPSD/CIMHub/blob/master/queries.txt>`__
The example below shows a query to obtain the correct mRID for switch SW2 in the IEEE 123 node model:
[ ]:
from gridappsd import topics as t
message = {
"modelId": model_mrid,
"requestType": "QUERY_OBJECT_DICT",
"resultFormat": "JSON",
"objectType": "LoadBreakSwitch"
}
response_obj = gapps.get_response(t.REQUEST_POWERGRID_DATA, message)
switch_dict = response_obj["data"]
# Filter to get mRID for switch SW2:
for index in switch_dict:
if index["IdentifiedObject.name"] == 'sw2':
sw_mrid = index["IdentifiedObject.mRID"]
Format of a Difference Message
The general format for a difference message is a python dictionary or equivalent JSON string that specifies the reverse difference and the forward difference, in compliance with the CIM standard:
The reverse difference is the current status / value associated with the control attribute. It is a formatted as a list of dictionary constructs, with each dictionary specifying the equipment mRID associated with the CIM class keys above, the control attribute, and the current value of that control attribute. The list can contain reverse differences for multiple pieces of equipment.
The forward difference is the desired new status / value associated with the control attribute. It is a formatted as a list of dictionary constructs, with each dictionary specifying the equipment mRID associated with the CIM class keys above, the control attribute, and the current value of that control attribute. The list can contain foward differences for multiple pieces of equipment.
message = {
"command": "update",
"input": {
"simulation_id": "simulation id as string",
"message": {
"timestamp": epoch time number,
"difference_mrid": "optional unique mRID for command logs",
"reverse_differences": [{
"object": "first equipment mRID",
"attribute": "control attribute",
"value": current value
},
{
"object": "second equipment mRID",
"attribute": "control attribute",
"value": current value
}
],
"forward_differences": [{
"object": "first equipment mRID",
"attribute": "control attribute",
"value": new value
},
{
"object": "second equipment mRID",
"attribute": "control attribute",
"value": new value
}
]
}
}
}
Note: The GridAPPS-D platform does not validate whether "reverse_differences": has the correct equipment control values for the current time. It is used just for compliance with the CIM standard.
Using GridAPPSD-Python DifferenceBuilder
The DifferenceBuilder class is a GridAPPSD-Python tool that can be used to automatically build the difference message with correct formatting.
First, import DifferenceBuilder from the GridAPPSD-Python Library and create an object that will be used to create the desired difference messages.
[ ]:
from gridappsd import DifferenceBuilder
my_diff_build = DifferenceBuilder(viz_simulation_id)
We then use two methods associated with the DifferenceBuilder object:
.add_difference(self, object_mrid, control_attribute, new_value, old_value)– Generates a correctly formatted difference message..get_message()– Saves the message as a python dictionary that can be published usinggapps.send(topic, message)
[ ]:
my_diff_build.add_difference(sw_mrid, "Switch.open", 1, 0) # Open switch given by sw_mrid
message = my_diff_build.get_message()
The difference message is then published to the GOSS Message Bus and the Simulation API using the .send() method associated with the GridAPPS-D connection object.
[ ]:
gapps.send(input_topic, message)
Unsubscribing from a Simulation
To unsubscribe from a simulation, pass the connection ID to .unsubscribe(conn_id) method of the GridAPPS-D connection object. The connection ID was obtained earlier when subscribing using conn_id = gapps.subscribe(topic, message).
[ ]:
gapps.unsubscribe(conn_id)