Heartbeat

An application to publish messages at regular time intervals, used for NOS-T stress tests

This application includes a CustomHeartbeat object class that mimics the time status messages of a managed application, but with an additional payload that includes a random string of specified length to vary the number of bytes per message and investigate it’s impact on delays.

class CustomHeartbeat(app, msg_length, time_status_step=None, time_status_init=None)

Bases: ScenarioTimeIntervalPublisher

Publishes time status messages at a regular interval (scenario time). The user can optionally provide a timedelta between time status messages and datetime for the initial time status message.

app

the application that will be publishing time status messages

Type:

Application

msg_length

number of characters per message, which also matches the number of bytes per message (1 char = 1 byte)

Type:

int

time_status_step

optional duration between time status ‘heartbeat’ messages (defaults to 1 second)

Type:

timedelta

time_status_init

optional initial scenario datetime to start publishing time status ‘heartbeat’ messages (defaults to simulation start time dictated from the manager)

Type:

datetime

CustomHeartbeat.publish_message()

Abstract publish_message method inherited from the ScenarioTimeIntervalPublisher object class from the publisher template in the NOS-T tools library

This method sends a message to the PREFIX/status/heartbeat/time which includes:

Parameters:

• name (str) – application name

• description (str) – optional application description (prints empty string if nothing provided)

• simTime (datetime) – current scenario datetime

• time (datetime) – current simulator wallclock datetime

• addPayload (str) – random string of user specified length

The following code demonstrates how the heartbeat application is started up and how the CustomHeartbeat ScenarioTimeIntervalPublisher object class is initialized and added to the simulator:

    # Note that these are loaded from a .env file in current working directory
    credentials = dotenv_values(".env")
    HOST, PORT = credentials["HOST"], int(credentials["PORT"])
    USERNAME, PASSWORD = credentials["USERNAME"], credentials["PASSWORD"]
    
    # set the client credentials
    config = ConnectionConfig(USERNAME, PASSWORD, HOST, PORT, True)

    # create the managed application
    app = ManagedApplication("heartbeat")
    
    # new heartbeat
    newBeat = CustomHeartbeat(app, MSG_LENGTH, timedelta(seconds=MSG_PERIOD))
    
    # add CustomHeartbeat as an observer
    app.simulator.add_observer(newBeat)

    # add a shutdown observer to shut down after a single test case
    app.simulator.add_observer(ShutDownObserver(app))
    
    # add ModeStatusObserver
    app.simulator.add_observer(ModeStatusObserver(app))

    # start up the application on PREFIX with 1 minute time steps, time status every 1 hour of scenario time
    app.start_up(
        PREFIX,
        config,
        True,
        time_step=timedelta(seconds=1) * SCALE,
        time_status_step=timedelta(seconds=24000000000),
        time_status_init=datetime(2021, 12, 14, 0, 0, tzinfo=timezone.utc)
    )
    
    # message sent to indicate message size and periodicity
    app.send_message(
        "settings",
        json.dumps({
            "messageBytes": str(MSG_LENGTH),
            "periodicity": str(MSG_PERIOD)
            }
        )
    )