Skip to content

nats_tools

ConfigGenerator

Source code in nats_tools/templates/config.py 
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
class ConfigGenerator:
    def __init__(self, template: t.Union[str, Path] = "default.conf.j2") -> None:
        """Create a new instance of config generator.

        Arguments:
            template: the template used to render configuration.
        """
        if isinstance(template, Path):
            self.template = load_template_from_path(template)
        elif Path(template).is_file():
            self.template = load_template_from_path(template)
        else:
            self.template = load_template_from_name(template)

    def render(
        self,
        address: str = "127.0.0.1",
        port: int = 4222,
        client_advertise: t.Optional[str] = None,
        server_name: t.Optional[str] = None,
        server_tags: t.Optional[t.Dict[str, str]] = None,
        user: t.Optional[str] = None,
        password: t.Optional[str] = None,
        users: t.Optional[t.List[t.Dict[str, t.Any]]] = None,
        token: t.Optional[str] = None,
        http_port: int = 8222,
        debug: t.Optional[bool] = None,
        trace: t.Optional[bool] = None,
        trace_verbose: t.Optional[bool] = None,
        logtime: t.Optional[bool] = None,
        pid_file: t.Union[str, Path, None] = None,
        port_file_dir: t.Union[str, Path, None] = None,
        log_file: t.Union[str, Path, None] = None,
        log_size_limit: t.Optional[int] = None,
        tls_cert: t.Union[str, Path, None] = None,
        tls_key: t.Union[str, Path, None] = None,
        tls_ca_cert: t.Union[str, Path, None] = None,
        cluster_name: t.Optional[str] = None,
        cluster_url: t.Optional[str] = None,
        cluster_listen: t.Optional[str] = None,
        routes: t.Optional[t.List[str]] = None,
        no_advertise: t.Optional[bool] = None,
        with_jetstream: bool = False,
        jetstream_domain: t.Optional[str] = None,
        store_directory: t.Union[str, Path, None] = None,
        max_memory_store: t.Optional[int] = None,
        max_file_store: t.Optional[int] = None,
        max_outstanding_catchup: t.Optional[int] = None,
        allow_leafnodes: bool = False,
        leafnodes_listen_address: t.Optional[str] = None,
        leafnodes_listen_port: t.Optional[int] = None,
        leafnode_remotes: t.Optional[t.Dict[str, t.Any]] = None,
        websocket_listen_address: t.Optional[str] = None,
        websocket_listen_port: t.Optional[int] = None,
        websocket_advertise_url: t.Optional[str] = None,
        websocket_tls: t.Optional[bool] = None,
        websocket_tls_cert: t.Union[str, Path, None] = None,
        websocket_tls_key: t.Union[str, Path, None] = None,
        websocket_same_origin: t.Optional[bool] = None,
        websocket_allowed_origins: t.Optional[t.List[str]] = None,
        websocket_compression: t.Optional[bool] = None,
        jwt_path: t.Union[str, Path, None] = None,
        operator: t.Optional[str] = None,
        system_account: t.Optional[str] = None,
        system_account_jwt: t.Optional[str] = None,
        allow_delete_jwt: t.Optional[bool] = None,
        compare_jwt_interval: t.Optional[str] = None,
        resolver_preload: t.Optional[t.Dict[str, str]] = None,
    ) -> str:
        """Render configuration according to arguments."""
        kwargs: t.Dict[str, t.Any] = {}

        kwargs["server_host"] = address
        kwargs["server_port"] = port
        kwargs["client_advertise"] = client_advertise
        kwargs["server_name"] = server_name
        kwargs["http_port"] = http_port

        if debug is not None:
            kwargs["debug"] = debug
        if trace is not None:
            kwargs["trace"] = trace
        if trace_verbose is not None:
            kwargs["trace_verbose"] = trace_verbose
        if logtime is not None:
            kwargs["logtime"] = logtime
        if pid_file is not None:
            kwargs["pid_file"] = Path(pid_file).as_posix()
        if port_file_dir is not None:
            kwargs["port_file_dir"] = Path(port_file_dir).as_posix()
        if log_file is not None:
            kwargs["log_file"] = Path(log_file).as_posix()
        if log_size_limit is not None:
            kwargs["log_size_limit"] = log_size_limit
        if server_tags:
            kwargs["server_tags"] = [
                f"{key}:{value}" for key, value in server_tags.items()
            ]

        cluster = False
        if cluster_listen or cluster_url:
            if cluster_listen is None:
                cluster_listen = cluster_url
            cluster = True
            kwargs["cluster_listen"] = cluster_listen
            if cluster_url is not None:
                kwargs["cluster_url"] = cluster_url
            if routes is not None:
                kwargs["routes"] = routes
            if no_advertise is not None:
                kwargs["no_advertise"] = no_advertise
            if cluster_name is not None:
                kwargs["cluster_name"] = cluster_name
        kwargs["cluster"] = cluster

        tls = False
        if tls_cert or tls_key:
            if not (tls_cert and tls_key):
                raise ValueError(
                    "tls_cert and tls_key argument must be provided together"
                )
            tls = True
            tls_cert_file = Path(tls_cert).as_posix()
            tls_key_file = Path(tls_key).as_posix()
            kwargs["tls_cert_file"] = tls_cert_file
            kwargs["tls_key_file"] = tls_key_file
            if tls_ca_cert:
                tls_ca_file = Path(tls_ca_cert).as_posix()
                kwargs["tls_ca_file"] = tls_ca_file
        kwargs["tls"] = tls
        kwargs["enable_jetstream"] = with_jetstream
        kwargs["jetstream_domain"] = jetstream_domain
        kwargs["max_file_store"] = max_file_store
        kwargs["max_memory_store"] = max_memory_store
        kwargs["max_outstanding_catchup"] = max_outstanding_catchup
        if store_directory is not None:
            kwargs["jetstream_store_dir"] = store_directory

        if user or password:
            if not (user and password):
                raise ValueError(
                    "Both user and password argument must be provided together"
                )

        if token:
            if user:
                raise ValueError(
                    "token argument cannot be used together with user and password"
                )

        if users:
            if token or user:
                raise ValueError(
                    "users argument cannot be used with token or user and password"
                )

        if operator:
            if users or token or user:
                raise ValueError(
                    "operator argument cannot be used with any of users, token, user and password arguments"
                )
            if system_account is None:
                raise ValueError("system_account argument must be provided")
            if system_account_jwt is None:
                raise ValueError("system_account_jwt argument must be provided")
            if jwt_path is None:
                raise ValueError("jwt_path argument must be provided")

        kwargs["user"] = user
        kwargs["password"] = password
        kwargs["users"] = users
        kwargs["token"] = token

        kwargs["operator"] = operator
        kwargs["system_account"] = system_account
        kwargs["jwt_path"] = jwt_path
        jwts = resolver_preload or {}
        if system_account and system_account_jwt:
            jwts[system_account] = system_account_jwt
        kwargs["jwts"] = jwts
        kwargs["allow_delete_jwt"] = allow_delete_jwt or False
        kwargs["compare_jwt_interval"] = compare_jwt_interval or "2m"

        if leafnodes_listen_address or leafnodes_listen_port:
            leafnodes_listen_address = leafnodes_listen_address or address
            leafnodes_listen_port = leafnodes_listen_port or 7422
            allow_leafnodes = True
            kwargs["leafnodes_listen_address"] = leafnodes_listen_address
            kwargs["leafnodes_listen_port"] = leafnodes_listen_port
        kwargs["allow_leafnodes"] = allow_leafnodes
        kwargs["leafnode_remotes"] = leafnode_remotes

        websocket = False
        if websocket_listen_port or websocket_listen_address:
            if websocket_listen_address is None:
                websocket_listen_address = address
            if websocket_listen_port is None:
                if websocket_tls or websocket_tls_cert:
                    websocket_listen_port = 443
            websocket = True
            kwargs["websocket_listen_port"] = websocket_listen_port
            kwargs["websocket_listen_address"] = websocket_listen_address
            if websocket_advertise_url:
                kwargs["websocket_advertise_url"] = websocket_advertise_url
            if websocket_tls_cert and websocket_tls_key:
                if not websocket_tls_cert and websocket_tls_key:
                    raise ValueError(
                        "websocket_tls_cert and websocket_tls_key must be provided to enable websocket TLS"
                    )
            if (
                (tls and websocket_tls) or (tls and websocket_tls is None)
            ) and websocket_tls_cert is None:
                if tls_cert is None or tls_key is None:
                    raise ValueError(
                        "websocket_tls_cert and websocket_tls_key must be provided to enable websocket TLS"
                    )
                websocket_tls_cert = Path(tls_cert).as_posix()
                websocket_tls_key = Path(tls_key).as_posix()
            websocket_tls = False
            if websocket_tls_cert:
                websocket_tls = True
                kwargs["websocket_tls_cert_file"] = websocket_tls_cert
                kwargs["websocket_tls_key_file"] = websocket_tls_key
            kwargs["websocket_tls"] = websocket_tls
            if websocket_tls:
                if websocket_same_origin is not None:
                    kwargs["websocket_same_origin"] = websocket_same_origin
                kwargs["websocket_allowed_origins"] = websocket_allowed_origins
            if websocket_compression is not None:
                kwargs["websocket_compression"] = websocket_compression
        kwargs["websocket"] = websocket

        return self.template.render(**kwargs)

__init__(template='default.conf.j2')

Create a new instance of config generator.

Parameters:

Name Type Description Default
template t.Union[str, Path]

the template used to render configuration.

 'default.conf.j2'
Source code in nats_tools/templates/config.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
def __init__(self, template: t.Union[str, Path] = "default.conf.j2") -> None:
    """Create a new instance of config generator.

    Arguments:
        template: the template used to render configuration.
    """
    if isinstance(template, Path):
        self.template = load_template_from_path(template)
    elif Path(template).is_file():
        self.template = load_template_from_path(template)
    else:
        self.template = load_template_from_name(template)

render(address='127.0.0.1', port=4222, client_advertise=None, server_name=None, server_tags=None, user=None, password=None, users=None, token=None, http_port=8222, debug=None, trace=None, trace_verbose=None, logtime=None, pid_file=None, port_file_dir=None, log_file=None, log_size_limit=None, tls_cert=None, tls_key=None, tls_ca_cert=None, cluster_name=None, cluster_url=None, cluster_listen=None, routes=None, no_advertise=None, with_jetstream=False, jetstream_domain=None, store_directory=None, max_memory_store=None, max_file_store=None, max_outstanding_catchup=None, allow_leafnodes=False, leafnodes_listen_address=None, leafnodes_listen_port=None, leafnode_remotes=None, websocket_listen_address=None, websocket_listen_port=None, websocket_advertise_url=None, websocket_tls=None, websocket_tls_cert=None, websocket_tls_key=None, websocket_same_origin=None, websocket_allowed_origins=None, websocket_compression=None, jwt_path=None, operator=None, system_account=None, system_account_jwt=None, allow_delete_jwt=None, compare_jwt_interval=None, resolver_preload=None)

Render configuration according to arguments.

Source code in nats_tools/templates/config.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def render(
    self,
    address: str = "127.0.0.1",
    port: int = 4222,
    client_advertise: t.Optional[str] = None,
    server_name: t.Optional[str] = None,
    server_tags: t.Optional[t.Dict[str, str]] = None,
    user: t.Optional[str] = None,
    password: t.Optional[str] = None,
    users: t.Optional[t.List[t.Dict[str, t.Any]]] = None,
    token: t.Optional[str] = None,
    http_port: int = 8222,
    debug: t.Optional[bool] = None,
    trace: t.Optional[bool] = None,
    trace_verbose: t.Optional[bool] = None,
    logtime: t.Optional[bool] = None,
    pid_file: t.Union[str, Path, None] = None,
    port_file_dir: t.Union[str, Path, None] = None,
    log_file: t.Union[str, Path, None] = None,
    log_size_limit: t.Optional[int] = None,
    tls_cert: t.Union[str, Path, None] = None,
    tls_key: t.Union[str, Path, None] = None,
    tls_ca_cert: t.Union[str, Path, None] = None,
    cluster_name: t.Optional[str] = None,
    cluster_url: t.Optional[str] = None,
    cluster_listen: t.Optional[str] = None,
    routes: t.Optional[t.List[str]] = None,
    no_advertise: t.Optional[bool] = None,
    with_jetstream: bool = False,
    jetstream_domain: t.Optional[str] = None,
    store_directory: t.Union[str, Path, None] = None,
    max_memory_store: t.Optional[int] = None,
    max_file_store: t.Optional[int] = None,
    max_outstanding_catchup: t.Optional[int] = None,
    allow_leafnodes: bool = False,
    leafnodes_listen_address: t.Optional[str] = None,
    leafnodes_listen_port: t.Optional[int] = None,
    leafnode_remotes: t.Optional[t.Dict[str, t.Any]] = None,
    websocket_listen_address: t.Optional[str] = None,
    websocket_listen_port: t.Optional[int] = None,
    websocket_advertise_url: t.Optional[str] = None,
    websocket_tls: t.Optional[bool] = None,
    websocket_tls_cert: t.Union[str, Path, None] = None,
    websocket_tls_key: t.Union[str, Path, None] = None,
    websocket_same_origin: t.Optional[bool] = None,
    websocket_allowed_origins: t.Optional[t.List[str]] = None,
    websocket_compression: t.Optional[bool] = None,
    jwt_path: t.Union[str, Path, None] = None,
    operator: t.Optional[str] = None,
    system_account: t.Optional[str] = None,
    system_account_jwt: t.Optional[str] = None,
    allow_delete_jwt: t.Optional[bool] = None,
    compare_jwt_interval: t.Optional[str] = None,
    resolver_preload: t.Optional[t.Dict[str, str]] = None,
) -> str:
    """Render configuration according to arguments."""
    kwargs: t.Dict[str, t.Any] = {}

    kwargs["server_host"] = address
    kwargs["server_port"] = port
    kwargs["client_advertise"] = client_advertise
    kwargs["server_name"] = server_name
    kwargs["http_port"] = http_port

    if debug is not None:
        kwargs["debug"] = debug
    if trace is not None:
        kwargs["trace"] = trace
    if trace_verbose is not None:
        kwargs["trace_verbose"] = trace_verbose
    if logtime is not None:
        kwargs["logtime"] = logtime
    if pid_file is not None:
        kwargs["pid_file"] = Path(pid_file).as_posix()
    if port_file_dir is not None:
        kwargs["port_file_dir"] = Path(port_file_dir).as_posix()
    if log_file is not None:
        kwargs["log_file"] = Path(log_file).as_posix()
    if log_size_limit is not None:
        kwargs["log_size_limit"] = log_size_limit
    if server_tags:
        kwargs["server_tags"] = [
            f"{key}:{value}" for key, value in server_tags.items()
        ]

    cluster = False
    if cluster_listen or cluster_url:
        if cluster_listen is None:
            cluster_listen = cluster_url
        cluster = True
        kwargs["cluster_listen"] = cluster_listen
        if cluster_url is not None:
            kwargs["cluster_url"] = cluster_url
        if routes is not None:
            kwargs["routes"] = routes
        if no_advertise is not None:
            kwargs["no_advertise"] = no_advertise
        if cluster_name is not None:
            kwargs["cluster_name"] = cluster_name
    kwargs["cluster"] = cluster

    tls = False
    if tls_cert or tls_key:
        if not (tls_cert and tls_key):
            raise ValueError(
                "tls_cert and tls_key argument must be provided together"
            )
        tls = True
        tls_cert_file = Path(tls_cert).as_posix()
        tls_key_file = Path(tls_key).as_posix()
        kwargs["tls_cert_file"] = tls_cert_file
        kwargs["tls_key_file"] = tls_key_file
        if tls_ca_cert:
            tls_ca_file = Path(tls_ca_cert).as_posix()
            kwargs["tls_ca_file"] = tls_ca_file
    kwargs["tls"] = tls
    kwargs["enable_jetstream"] = with_jetstream
    kwargs["jetstream_domain"] = jetstream_domain
    kwargs["max_file_store"] = max_file_store
    kwargs["max_memory_store"] = max_memory_store
    kwargs["max_outstanding_catchup"] = max_outstanding_catchup
    if store_directory is not None:
        kwargs["jetstream_store_dir"] = store_directory

    if user or password:
        if not (user and password):
            raise ValueError(
                "Both user and password argument must be provided together"
            )

    if token:
        if user:
            raise ValueError(
                "token argument cannot be used together with user and password"
            )

    if users:
        if token or user:
            raise ValueError(
                "users argument cannot be used with token or user and password"
            )

    if operator:
        if users or token or user:
            raise ValueError(
                "operator argument cannot be used with any of users, token, user and password arguments"
            )
        if system_account is None:
            raise ValueError("system_account argument must be provided")
        if system_account_jwt is None:
            raise ValueError("system_account_jwt argument must be provided")
        if jwt_path is None:
            raise ValueError("jwt_path argument must be provided")

    kwargs["user"] = user
    kwargs["password"] = password
    kwargs["users"] = users
    kwargs["token"] = token

    kwargs["operator"] = operator
    kwargs["system_account"] = system_account
    kwargs["jwt_path"] = jwt_path
    jwts = resolver_preload or {}
    if system_account and system_account_jwt:
        jwts[system_account] = system_account_jwt
    kwargs["jwts"] = jwts
    kwargs["allow_delete_jwt"] = allow_delete_jwt or False
    kwargs["compare_jwt_interval"] = compare_jwt_interval or "2m"

    if leafnodes_listen_address or leafnodes_listen_port:
        leafnodes_listen_address = leafnodes_listen_address or address
        leafnodes_listen_port = leafnodes_listen_port or 7422
        allow_leafnodes = True
        kwargs["leafnodes_listen_address"] = leafnodes_listen_address
        kwargs["leafnodes_listen_port"] = leafnodes_listen_port
    kwargs["allow_leafnodes"] = allow_leafnodes
    kwargs["leafnode_remotes"] = leafnode_remotes

    websocket = False
    if websocket_listen_port or websocket_listen_address:
        if websocket_listen_address is None:
            websocket_listen_address = address
        if websocket_listen_port is None:
            if websocket_tls or websocket_tls_cert:
                websocket_listen_port = 443
        websocket = True
        kwargs["websocket_listen_port"] = websocket_listen_port
        kwargs["websocket_listen_address"] = websocket_listen_address
        if websocket_advertise_url:
            kwargs["websocket_advertise_url"] = websocket_advertise_url
        if websocket_tls_cert and websocket_tls_key:
            if not websocket_tls_cert and websocket_tls_key:
                raise ValueError(
                    "websocket_tls_cert and websocket_tls_key must be provided to enable websocket TLS"
                )
        if (
            (tls and websocket_tls) or (tls and websocket_tls is None)
        ) and websocket_tls_cert is None:
            if tls_cert is None or tls_key is None:
                raise ValueError(
                    "websocket_tls_cert and websocket_tls_key must be provided to enable websocket TLS"
                )
            websocket_tls_cert = Path(tls_cert).as_posix()
            websocket_tls_key = Path(tls_key).as_posix()
        websocket_tls = False
        if websocket_tls_cert:
            websocket_tls = True
            kwargs["websocket_tls_cert_file"] = websocket_tls_cert
            kwargs["websocket_tls_key_file"] = websocket_tls_key
        kwargs["websocket_tls"] = websocket_tls
        if websocket_tls:
            if websocket_same_origin is not None:
                kwargs["websocket_same_origin"] = websocket_same_origin
            kwargs["websocket_allowed_origins"] = websocket_allowed_origins
        if websocket_compression is not None:
            kwargs["websocket_compression"] = websocket_compression
    kwargs["websocket"] = websocket

    return self.template.render(**kwargs)

NATSD

Source code in nats_tools/natsd.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
class NATSD:
    def __init__(
        self,
        address: str = "127.0.0.1",
        port: int = 4222,
        client_advertise: t.Optional[str] = None,
        server_name: t.Optional[str] = None,
        server_tags: t.Optional[t.Dict[str, str]] = None,
        user: t.Optional[str] = None,
        password: t.Optional[str] = None,
        users: t.Optional[t.List[t.Dict[str, t.Any]]] = None,
        token: t.Optional[str] = None,
        http_port: int = 8222,
        debug: t.Optional[bool] = None,
        trace: t.Optional[bool] = None,
        trace_verbose: t.Optional[bool] = None,
        logtime: t.Optional[bool] = None,
        pid_file: t.Union[str, Path, None] = None,
        port_file_dir: t.Union[str, Path, None] = None,
        log_file: t.Union[str, Path, None] = None,
        log_size_limit: t.Optional[int] = None,
        tls_cert: t.Union[str, Path, None] = None,
        tls_key: t.Union[str, Path, None] = None,
        tls_ca_cert: t.Union[str, Path, None] = None,
        cluster_name: t.Optional[str] = None,
        cluster_url: t.Optional[str] = None,
        cluster_listen: t.Optional[str] = None,
        routes: t.Optional[t.List[str]] = None,
        no_advertise: t.Optional[bool] = None,
        with_jetstream: bool = False,
        jetstream_domain: t.Optional[str] = None,
        store_directory: t.Union[str, Path, None] = None,
        max_memory_store: t.Optional[int] = None,
        max_file_store: t.Optional[int] = None,
        max_outstanding_catchup: t.Optional[int] = None,
        allow_leafnodes: bool = False,
        leafnodes_listen_address: t.Optional[str] = None,
        leafnodes_listen_port: t.Optional[int] = None,
        leafnode_remotes: t.Optional[t.Dict[str, t.Any]] = None,
        websocket_listen_address: t.Optional[str] = None,
        websocket_listen_port: t.Optional[int] = None,
        websocket_advertise_url: t.Optional[str] = None,
        websocket_tls: t.Optional[bool] = None,
        websocket_tls_cert: t.Union[str, Path, None] = None,
        websocket_tls_key: t.Union[str, Path, None] = None,
        websocket_same_origin: t.Optional[bool] = None,
        websocket_allowed_origins: t.Optional[t.List[str]] = None,
        websocket_compression: t.Optional[bool] = None,
        jwt_path: t.Union[str, Path, None] = None,
        operator: t.Optional[str] = None,
        system_account: t.Optional[str] = None,
        system_account_jwt: t.Optional[str] = None,
        allow_delete_jwt: t.Optional[bool] = None,
        compare_jwt_interval: t.Optional[str] = None,
        resolver_preload: t.Optional[t.Dict[str, str]] = None,
        config_file: t.Union[str, Path, None] = None,
        max_cpus: t.Optional[float] = None,
        start_timeout: float = 1,
    ) -> None:
        """Create a new instance of nats-server daemon.

        Arguments:
            address: host address nats-server should listen to. Default is 127.0.0.1 (localhost).
            port: tcp port nats-server should listen to. Clients can connect to this port. Default is 4222.
            server_name: the server name. Default to auto-generated name.
            user: username required for connections. Omitted by default.
            password: password required for connections. Omitted by default.
            token: authorization token required for connections. Omitted by default.
            http_port: port for http monitoring. Default is 8222.
            debug: enable debugging output. Default is False.
            trace: enable raw traces. Default is False.
            pid_file: file to write process ID to. Omitted by default.
            log_file: file to redirect log output to. Omitted by default.
            tls_cert: server certificate file (TLS is enabled when both cert and key are provided)
            tls_key: server key file (TLS is enabled when both cert and key are provided)
            tls_ca_cert: client certificate for CA verification (mutual TLS is enabled when ca cert is provided)
            cluster_name: the cluster name. Default to auto-generated name when clustering is enabled.
            cluster_url: cluster URL for sollicited routes.
            cluster_listen: cluster URL from which members can solicite routes. Enable cluster mode when set.
            routes: routes to solicit and connect.
            no_advertise: do not advertise known cluster information to clients.
            with_jetstream: enable jetstream engine when True. Disabled by default.
            store_directory: path to jetstream store directory. Default to a temporary directory.
            config_file: path to a configuration file. None by default.
            max_cpus: maximum number of CPU configured using GOMAXPROCS environment variable. By default all CPUs can be used.
            start_timeout: amount of time to wait before raising an error when starting the daemon with wait=True.
        """
        if config_file is None:
            config_file = Path(tempfile.mkdtemp()).joinpath("nats.conf")
            generator = ConfigGenerator()
            config_str = generator.render(
                address=address,
                port=port,
                client_advertise=client_advertise,
                server_name=server_name,
                server_tags=server_tags,
                user=user,
                password=password,
                users=users,
                token=token,
                http_port=http_port,
                debug=debug,
                trace=trace,
                trace_verbose=trace_verbose,
                logtime=logtime,
                pid_file=pid_file,
                port_file_dir=port_file_dir,
                log_file=log_file,
                log_size_limit=log_size_limit,
                tls_cert=tls_cert,
                tls_key=tls_key,
                tls_ca_cert=tls_ca_cert,
                cluster_name=cluster_name,
                cluster_url=cluster_url,
                cluster_listen=cluster_listen,
                routes=routes,
                no_advertise=no_advertise,
                with_jetstream=with_jetstream,
                jetstream_domain=jetstream_domain,
                store_directory=store_directory,
                max_memory_store=max_memory_store,
                max_file_store=max_file_store,
                max_outstanding_catchup=max_outstanding_catchup,
                allow_leafnodes=allow_leafnodes,
                leafnodes_listen_address=leafnodes_listen_address,
                leafnodes_listen_port=leafnodes_listen_port,
                leafnode_remotes=leafnode_remotes,
                websocket_listen_address=websocket_listen_address,
                websocket_listen_port=websocket_listen_port,
                websocket_advertise_url=websocket_advertise_url,
                websocket_tls=websocket_tls,
                websocket_tls_cert=websocket_tls_cert,
                websocket_tls_key=websocket_tls_key,
                websocket_same_origin=websocket_same_origin,
                websocket_allowed_origins=websocket_allowed_origins,
                websocket_compression=websocket_compression,
                jwt_path=jwt_path,
                operator=operator,
                system_account=system_account,
                system_account_jwt=system_account_jwt,
                allow_delete_jwt=allow_delete_jwt,
                compare_jwt_interval=compare_jwt_interval,
                resolver_preload=resolver_preload,
            )
            config_file.write_text(config_str)
            weakref.finalize(self, shutil.rmtree, config_file.parent, True)
        self.server_name = server_name
        self.address = address
        self.port = port
        self.user = user
        self.password = password
        self.timeout = start_timeout
        self.http_port = http_port
        self.token = token
        self.bin_name = "nats-server"
        self.bin_path: t.Optional[str] = None
        self.config_file = Path(config_file) if config_file else None
        self.debug = debug or os.environ.get("DEBUG_NATS_TEST", "") in (
            "true",
            "1",
            "y",
            "yes",
            "on",
        )
        self.trace = trace or os.environ.get("DEBUG_NATS_TEST", "") in (
            "true",
            "1",
            "y",
            "yes",
            "on",
        )
        self.pid_file = Path(pid_file).absolute().as_posix() if pid_file else None
        self.log_file = Path(log_file).absolute().as_posix() if log_file else None
        self.max_cpus = max_cpus

        self.tls_cert = tls_cert
        self.tls_key = tls_key
        self.tls_ca_cert = tls_ca_cert
        if self.tls_ca_cert and self.tls_cert and self.tls_key:
            self.tls_verify = True
            self.tls = False
        elif self.tls_cert and self.tls_key:
            self.tls_verify = False
            self.tls = True
        elif self.tls_ca_cert:
            raise ValueError(
                "Both certificate and key files must be provided with a CA certificate"
            )
        elif self.tls_cert or self.tls_key:
            raise ValueError("Both certificate and key files must be provided")
        else:
            self.tls = False
            self.tls_verify = False

        self.cluster_name = cluster_name
        self.cluster_url = cluster_url
        self.cluster_listen = cluster_listen
        self.routes = routes
        self.no_advertise = no_advertise

        self.jetstream_enabled = with_jetstream
        if store_directory:
            self.store_dir = Path(store_directory)
            self._store_dir_is_temporary = False
        else:
            self.store_dir = Path(tempfile.mkdtemp()).resolve(True)
            self._store_dir_is_temporary = True
            weakref.finalize(self, shutil.rmtree, self.store_dir.as_posix(), True)

        self.proc: t.Optional["subprocess.Popen[bytes]"] = None
        self.monitor = NATSMonitor(f"http://{self.address}:{self.http_port}")

    def is_alive(self) -> bool:
        if self.proc is None:
            return False
        return self.proc.poll() is None

    def _cleanup_on_exit(self) -> None:
        if self.proc and self.proc.poll() is None:
            print(
                "[\033[0;31mWARNING\033[0;0m] Stopping server listening on %d."
                % self.port
            )
            self.kill()

    def start(self, wait: bool = False) -> "NATSD":
        # Check if there is an nats-server binary in the current working directory
        if Path(self.bin_name).is_file():
            self.bin_path = Path(self.bin_name).resolve(True).as_posix()
        # Path in `../scripts/install_nats.sh`
        elif DEFAULT_BIN_DIR.joinpath(self.bin_name).is_file():
            self.bin_path = DEFAULT_BIN_DIR.joinpath(self.bin_name).as_posix()
        # This directory contains binary
        else:
            self.bin_path = shutil.which(self.bin_name)
            if self.bin_path is None:
                raise FileNotFoundError("nats-server executable not found")

        cmd = [
            self.bin_path,
            "-p",
            "%d" % self.port,
            "-m",
            "%d" % self.http_port,
            "-a",
            self.address,
        ]

        if self.config_file is not None:
            if not self.config_file.exists():
                raise FileNotFoundError(self.config_file)
            else:
                config_file = self.config_file.absolute().as_posix()
            cmd.append("--config")
            cmd.append(config_file)

        env = os.environ.copy()

        if self.max_cpus:
            env["GOMAXPROCS"] = format(self.max_cpus, ".2f")

        if self.debug:
            self.proc = subprocess.Popen(cmd, env=env)
        else:
            self.proc = subprocess.Popen(
                cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env
            )

        if self.debug:
            print(
                "[\033[0;33mDEBUG\033[0;0m] Server listening on port %d started."
                % self.port
            )
        if wait:
            deadline = time.time() + self.timeout or float("inf")
            while True:
                status = self.proc.poll()
                if status is not None:
                    if self.debug:
                        print(
                            "[\033[0;31mWARNING\033[0;0m] Server listening on port {port} already finished running with exit {ret}".format(
                                port=self.port, ret=self.proc.returncode
                            )
                        )
                    raise subprocess.CalledProcessError(
                        returncode=self.proc.returncode, cmd=self.proc.args
                    )
                if time.time() > deadline:
                    self.stop()
                    raise TimeoutError(
                        f"nats-server failed to start before timeout ({self.timeout:.3f}s)"
                    )
                try:
                    self.monitor.varz()
                    break
                except httpx.HTTPError as exc:
                    print(
                        f"[\033[0;31mDEBUG\033[0;0m] Waiting for server to be up. Last error: {type(exc).__name__} - {repr(exc)}."
                    )
                    time.sleep(0.1)
                    continue

        weakref.finalize(self, self._cleanup_on_exit)
        return self

    def stop(self, timeout: t.Optional[float] = 10) -> None:
        if self.debug:
            print(
                "[\033[0;33mDEBUG\033[0;0m] Server listening on %d will stop."
                % self.port
            )

        if self.proc is None:
            if self.debug:
                print(
                    "[\033[0;31mWARNING\033[0;0m] Failed terminating server listening on port %d"
                    % self.port
                )

        elif self.proc.returncode is not None:
            if self.debug:
                print(
                    "[\033[0;31mWARNING\033[0;0m] Server listening on port {port} already finished running with exit {ret}".format(
                        port=self.port, ret=self.proc.returncode
                    )
                )
        else:
            try:
                self.term(timeout=timeout)
            except TimeoutError:
                self.kill()
            if self.debug:
                print(
                    "[\033[0;33mDEBUG\033[0;0m] Server listening on %d was stopped."
                    % self.port
                )
        expected = 15 if os.name == "nt" else 1
        if self.proc and self.proc.returncode != expected:
            raise subprocess.CalledProcessError(
                self.proc.returncode, cmd=self.proc.args
            )

    def wait(self, timeout: t.Optional[float] = None) -> int:
        """Wait for process to finish and return status code.

        Possible status codes (non-exhaustive):
            -1: process is not started yet.
            0: process has been stopped after entering lame duck mode.
            1: process has been stopped due to TERM signal.
            2: process has been stopped due to QUIT signal.
            -9: process has been stopped due to KILL signal.
        """
        if self.proc is None:
            return 0
        status = self.proc.poll()
        if status is not None:
            return status
        return self.proc.wait(timeout=timeout)

    def send_signal(self, sig: t.Union[int, signal.Signals, Signal]) -> None:
        if self.proc is None:
            raise TypeError("Process is not started yet")
        status = self.proc.poll()
        if status is not None:
            raise subprocess.CalledProcessError(status, cmd=self.proc.args)
        if os.name != "nt":
            if not isinstance(sig, Signal):
                sig = signal.Signals(sig)
                sig = Signal(sig)
            os.kill(self.proc.pid, sig.value)
        else:
            sig = Signal(sig)
            if isinstance(sig.value, InvalidWindowsSignal):
                # Use a subprocess to explicitely call `nats-server --signal` which will handle signal correctly on Windows
                if sig.value == InvalidWindowsSignal.SIGKILL:
                    os.kill(self.proc.pid, signal.SIGINT)
                elif sig.value == InvalidWindowsSignal.SIGQUIT:
                    os.kill(self.proc.pid, signal.SIGBREAK)  # type: ignore[attr-defined]
                elif sig.value == InvalidWindowsSignal.SIGHUP:
                    warnings.warn("Config reload is not supported on Windows")
                elif sig.value == InvalidWindowsSignal.SIGUSR1:
                    warnings.warn("Log file roration is not supported on Windows")
                elif sig.value == InvalidWindowsSignal.SIGUSR2:
                    warnings.warn("Lame Duck Mode is not supported on Windows")
                    os.kill(self.proc.pid, signal.SIGINT)
            else:
                os.kill(self.proc.pid, sig.value)

    def quit(self, timeout: t.Optional[float] = None) -> None:
        self.send_signal(Signal.QUIT)
        self.wait(timeout=timeout)

    def kill(self, timeout: t.Optional[float] = None) -> None:
        self.send_signal(Signal.KILL)
        self.wait(timeout=timeout)

    def term(self, timeout: t.Optional[float] = 10) -> None:
        self.send_signal(Signal.STOP)
        self.wait(timeout=timeout)

    def reopen_log_file(self) -> None:
        self.send_signal(Signal.REOPEN)

    def enter_lame_duck_mode(self) -> None:
        self.send_signal(Signal.LDM)

    def reload_config(self) -> None:
        self.send_signal(Signal.RELOAD)

    def __enter__(self) -> "NATSD":
        return self.start(wait=True)

    def __exit__(
        self,
        error_type: t.Optional[t.Type[BaseException]] = None,
        error: t.Optional[BaseException] = None,
        traceback: t.Optional[types.TracebackType] = None,
    ) -> None:
        self.stop()

__init__(address='127.0.0.1', port=4222, client_advertise=None, server_name=None, server_tags=None, user=None, password=None, users=None, token=None, http_port=8222, debug=None, trace=None, trace_verbose=None, logtime=None, pid_file=None, port_file_dir=None, log_file=None, log_size_limit=None, tls_cert=None, tls_key=None, tls_ca_cert=None, cluster_name=None, cluster_url=None, cluster_listen=None, routes=None, no_advertise=None, with_jetstream=False, jetstream_domain=None, store_directory=None, max_memory_store=None, max_file_store=None, max_outstanding_catchup=None, allow_leafnodes=False, leafnodes_listen_address=None, leafnodes_listen_port=None, leafnode_remotes=None, websocket_listen_address=None, websocket_listen_port=None, websocket_advertise_url=None, websocket_tls=None, websocket_tls_cert=None, websocket_tls_key=None, websocket_same_origin=None, websocket_allowed_origins=None, websocket_compression=None, jwt_path=None, operator=None, system_account=None, system_account_jwt=None, allow_delete_jwt=None, compare_jwt_interval=None, resolver_preload=None, config_file=None, max_cpus=None, start_timeout=1)

Create a new instance of nats-server daemon.

Parameters:

Name Type Description Default
address str

host address nats-server should listen to. Default is 127.0.0.1 (localhost).

 '127.0.0.1'
port int

tcp port nats-server should listen to. Clients can connect to this port. Default is 4222.

 4222
server_name t.Optional[str]

the server name. Default to auto-generated name.

 None
user t.Optional[str]

username required for connections. Omitted by default.

 None
password t.Optional[str]

password required for connections. Omitted by default.

 None
token t.Optional[str]

authorization token required for connections. Omitted by default.

 None
http_port int

port for http monitoring. Default is 8222.

 8222
debug t.Optional[bool]

enable debugging output. Default is False.

 None
trace t.Optional[bool]

enable raw traces. Default is False.

 None
pid_file t.Union[str, Path, None]

file to write process ID to. Omitted by default.

 None
log_file t.Union[str, Path, None]

file to redirect log output to. Omitted by default.

 None
tls_cert t.Union[str, Path, None]

server certificate file (TLS is enabled when both cert and key are provided)

 None
tls_key t.Union[str, Path, None]

server key file (TLS is enabled when both cert and key are provided)

 None
tls_ca_cert t.Union[str, Path, None]

client certificate for CA verification (mutual TLS is enabled when ca cert is provided)

 None
cluster_name t.Optional[str]

the cluster name. Default to auto-generated name when clustering is enabled.

 None
cluster_url t.Optional[str]

cluster URL for sollicited routes.

 None
cluster_listen t.Optional[str]

cluster URL from which members can solicite routes. Enable cluster mode when set.

 None
routes t.Optional[t.List[str]]

routes to solicit and connect.

 None
no_advertise t.Optional[bool]

do not advertise known cluster information to clients.

 None
with_jetstream bool

enable jetstream engine when True. Disabled by default.

 False
store_directory t.Union[str, Path, None]

path to jetstream store directory. Default to a temporary directory.

 None
config_file t.Union[str, Path, None]

path to a configuration file. None by default.

 None
max_cpus t.Optional[float]

maximum number of CPU configured using GOMAXPROCS environment variable. By default all CPUs can be used.

 None
start_timeout float

amount of time to wait before raising an error when starting the daemon with wait=True.

 1
Source code in nats_tools/natsd.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def __init__(
    self,
    address: str = "127.0.0.1",
    port: int = 4222,
    client_advertise: t.Optional[str] = None,
    server_name: t.Optional[str] = None,
    server_tags: t.Optional[t.Dict[str, str]] = None,
    user: t.Optional[str] = None,
    password: t.Optional[str] = None,
    users: t.Optional[t.List[t.Dict[str, t.Any]]] = None,
    token: t.Optional[str] = None,
    http_port: int = 8222,
    debug: t.Optional[bool] = None,
    trace: t.Optional[bool] = None,
    trace_verbose: t.Optional[bool] = None,
    logtime: t.Optional[bool] = None,
    pid_file: t.Union[str, Path, None] = None,
    port_file_dir: t.Union[str, Path, None] = None,
    log_file: t.Union[str, Path, None] = None,
    log_size_limit: t.Optional[int] = None,
    tls_cert: t.Union[str, Path, None] = None,
    tls_key: t.Union[str, Path, None] = None,
    tls_ca_cert: t.Union[str, Path, None] = None,
    cluster_name: t.Optional[str] = None,
    cluster_url: t.Optional[str] = None,
    cluster_listen: t.Optional[str] = None,
    routes: t.Optional[t.List[str]] = None,
    no_advertise: t.Optional[bool] = None,
    with_jetstream: bool = False,
    jetstream_domain: t.Optional[str] = None,
    store_directory: t.Union[str, Path, None] = None,
    max_memory_store: t.Optional[int] = None,
    max_file_store: t.Optional[int] = None,
    max_outstanding_catchup: t.Optional[int] = None,
    allow_leafnodes: bool = False,
    leafnodes_listen_address: t.Optional[str] = None,
    leafnodes_listen_port: t.Optional[int] = None,
    leafnode_remotes: t.Optional[t.Dict[str, t.Any]] = None,
    websocket_listen_address: t.Optional[str] = None,
    websocket_listen_port: t.Optional[int] = None,
    websocket_advertise_url: t.Optional[str] = None,
    websocket_tls: t.Optional[bool] = None,
    websocket_tls_cert: t.Union[str, Path, None] = None,
    websocket_tls_key: t.Union[str, Path, None] = None,
    websocket_same_origin: t.Optional[bool] = None,
    websocket_allowed_origins: t.Optional[t.List[str]] = None,
    websocket_compression: t.Optional[bool] = None,
    jwt_path: t.Union[str, Path, None] = None,
    operator: t.Optional[str] = None,
    system_account: t.Optional[str] = None,
    system_account_jwt: t.Optional[str] = None,
    allow_delete_jwt: t.Optional[bool] = None,
    compare_jwt_interval: t.Optional[str] = None,
    resolver_preload: t.Optional[t.Dict[str, str]] = None,
    config_file: t.Union[str, Path, None] = None,
    max_cpus: t.Optional[float] = None,
    start_timeout: float = 1,
) -> None:
    """Create a new instance of nats-server daemon.

    Arguments:
        address: host address nats-server should listen to. Default is 127.0.0.1 (localhost).
        port: tcp port nats-server should listen to. Clients can connect to this port. Default is 4222.
        server_name: the server name. Default to auto-generated name.
        user: username required for connections. Omitted by default.
        password: password required for connections. Omitted by default.
        token: authorization token required for connections. Omitted by default.
        http_port: port for http monitoring. Default is 8222.
        debug: enable debugging output. Default is False.
        trace: enable raw traces. Default is False.
        pid_file: file to write process ID to. Omitted by default.
        log_file: file to redirect log output to. Omitted by default.
        tls_cert: server certificate file (TLS is enabled when both cert and key are provided)
        tls_key: server key file (TLS is enabled when both cert and key are provided)
        tls_ca_cert: client certificate for CA verification (mutual TLS is enabled when ca cert is provided)
        cluster_name: the cluster name. Default to auto-generated name when clustering is enabled.
        cluster_url: cluster URL for sollicited routes.
        cluster_listen: cluster URL from which members can solicite routes. Enable cluster mode when set.
        routes: routes to solicit and connect.
        no_advertise: do not advertise known cluster information to clients.
        with_jetstream: enable jetstream engine when True. Disabled by default.
        store_directory: path to jetstream store directory. Default to a temporary directory.
        config_file: path to a configuration file. None by default.
        max_cpus: maximum number of CPU configured using GOMAXPROCS environment variable. By default all CPUs can be used.
        start_timeout: amount of time to wait before raising an error when starting the daemon with wait=True.
    """
    if config_file is None:
        config_file = Path(tempfile.mkdtemp()).joinpath("nats.conf")
        generator = ConfigGenerator()
        config_str = generator.render(
            address=address,
            port=port,
            client_advertise=client_advertise,
            server_name=server_name,
            server_tags=server_tags,
            user=user,
            password=password,
            users=users,
            token=token,
            http_port=http_port,
            debug=debug,
            trace=trace,
            trace_verbose=trace_verbose,
            logtime=logtime,
            pid_file=pid_file,
            port_file_dir=port_file_dir,
            log_file=log_file,
            log_size_limit=log_size_limit,
            tls_cert=tls_cert,
            tls_key=tls_key,
            tls_ca_cert=tls_ca_cert,
            cluster_name=cluster_name,
            cluster_url=cluster_url,
            cluster_listen=cluster_listen,
            routes=routes,
            no_advertise=no_advertise,
            with_jetstream=with_jetstream,
            jetstream_domain=jetstream_domain,
            store_directory=store_directory,
            max_memory_store=max_memory_store,
            max_file_store=max_file_store,
            max_outstanding_catchup=max_outstanding_catchup,
            allow_leafnodes=allow_leafnodes,
            leafnodes_listen_address=leafnodes_listen_address,
            leafnodes_listen_port=leafnodes_listen_port,
            leafnode_remotes=leafnode_remotes,
            websocket_listen_address=websocket_listen_address,
            websocket_listen_port=websocket_listen_port,
            websocket_advertise_url=websocket_advertise_url,
            websocket_tls=websocket_tls,
            websocket_tls_cert=websocket_tls_cert,
            websocket_tls_key=websocket_tls_key,
            websocket_same_origin=websocket_same_origin,
            websocket_allowed_origins=websocket_allowed_origins,
            websocket_compression=websocket_compression,
            jwt_path=jwt_path,
            operator=operator,
            system_account=system_account,
            system_account_jwt=system_account_jwt,
            allow_delete_jwt=allow_delete_jwt,
            compare_jwt_interval=compare_jwt_interval,
            resolver_preload=resolver_preload,
        )
        config_file.write_text(config_str)
        weakref.finalize(self, shutil.rmtree, config_file.parent, True)
    self.server_name = server_name
    self.address = address
    self.port = port
    self.user = user
    self.password = password
    self.timeout = start_timeout
    self.http_port = http_port
    self.token = token
    self.bin_name = "nats-server"
    self.bin_path: t.Optional[str] = None
    self.config_file = Path(config_file) if config_file else None
    self.debug = debug or os.environ.get("DEBUG_NATS_TEST", "") in (
        "true",
        "1",
        "y",
        "yes",
        "on",
    )
    self.trace = trace or os.environ.get("DEBUG_NATS_TEST", "") in (
        "true",
        "1",
        "y",
        "yes",
        "on",
    )
    self.pid_file = Path(pid_file).absolute().as_posix() if pid_file else None
    self.log_file = Path(log_file).absolute().as_posix() if log_file else None
    self.max_cpus = max_cpus

    self.tls_cert = tls_cert
    self.tls_key = tls_key
    self.tls_ca_cert = tls_ca_cert
    if self.tls_ca_cert and self.tls_cert and self.tls_key:
        self.tls_verify = True
        self.tls = False
    elif self.tls_cert and self.tls_key:
        self.tls_verify = False
        self.tls = True
    elif self.tls_ca_cert:
        raise ValueError(
            "Both certificate and key files must be provided with a CA certificate"
        )
    elif self.tls_cert or self.tls_key:
        raise ValueError("Both certificate and key files must be provided")
    else:
        self.tls = False
        self.tls_verify = False

    self.cluster_name = cluster_name
    self.cluster_url = cluster_url
    self.cluster_listen = cluster_listen
    self.routes = routes
    self.no_advertise = no_advertise

    self.jetstream_enabled = with_jetstream
    if store_directory:
        self.store_dir = Path(store_directory)
        self._store_dir_is_temporary = False
    else:
        self.store_dir = Path(tempfile.mkdtemp()).resolve(True)
        self._store_dir_is_temporary = True
        weakref.finalize(self, shutil.rmtree, self.store_dir.as_posix(), True)

    self.proc: t.Optional["subprocess.Popen[bytes]"] = None
    self.monitor = NATSMonitor(f"http://{self.address}:{self.http_port}")

wait(timeout=None)

Wait for process to finish and return status code.

Possible status codes (non-exhaustive): -1: process is not started yet. 0: process has been stopped after entering lame duck mode. 1: process has been stopped due to TERM signal. 2: process has been stopped due to QUIT signal. -9: process has been stopped due to KILL signal.

Source code in nats_tools/natsd.py 
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
def wait(self, timeout: t.Optional[float] = None) -> int:
    """Wait for process to finish and return status code.

    Possible status codes (non-exhaustive):
        -1: process is not started yet.
        0: process has been stopped after entering lame duck mode.
        1: process has been stopped due to TERM signal.
        2: process has been stopped due to QUIT signal.
        -9: process has been stopped due to KILL signal.
    """
    if self.proc is None:
        return 0
    status = self.proc.poll()
    if status is not None:
        return status
    return self.proc.wait(timeout=timeout)

NATSMonitor

Source code in nats_tools/monitor.py 
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
class NATSMonitor:
    def __init__(self, endpoint: str) -> None:
        self.endpoint = endpoint
        self._client: t.Optional[httpx.Client] = None

    def _request(self, endpoint: str, **params: t.Any) -> t.Dict[str, t.Any]:
        if self._client is None:
            self._client = httpx.Client(base_url=self.endpoint)
        response = self._client.get(endpoint, params=params)
        response.raise_for_status()
        return t.cast(t.Dict[str, t.Any], response.json())

    def varz(self) -> t.Dict[str, t.Any]:
        """The /varz endpoint returns general information about the server state and configuration.

        Example: https://demo.nats.io:8222/varz
        """
        return self._request("/varz")

    def jsz(
        self,
        acc: t.Optional[str] = None,
        accounts: t.Optional[bool] = None,
        streams: t.Optional[bool] = None,
        consumers: t.Optional[bool] = None,
        config: t.Optional[bool] = None,
        leader_only: bool = False,
        offset: int = 0,
        limit: int = 1024,
    ) -> t.Dict[str, t.Any]:
        """The /jsz endpoint reports more detailed information on JetStream.

        For accounts, it uses a paging mechanism that defaults to 1024 connections.

        NOTE: If you're in a clustered environment, it is recommended to retrieve the information
              from the stream's leader in order to get the most accurate and up-to-date data.

        Arguments:
            acc: include metrics for the specified account only. Omitted by default.
            accounts: include account specific jetstream information. Default is False.
            streams: include streams. When set, implies `accounts=True`. Default is False.
            consumers: include consumers. When set, implies `stream=True`. Default is False.
            config: when stream or consumer are requested, include their respective configuration. Default is False.
            leader_only: only the leader responds. Default is False.
            offset: pagination offset. Default is 0.
            limit: number of results to return. Default is 1024.

        Returns:
            results as a dictionary.
        """
        params: t.Dict[str, t.Any] = {
            "leader-only": leader_only,
            "limit": limit,
            "offset": offset,
        }
        if accounts:
            params["accounts"] = accounts
        if streams:
            params["streams"] = streams
        if consumers:
            params["consumers"] = consumers
        if config:
            params["config"] = config
        if acc:
            params["acc"] = acc
        return self._request("/jsz", **params)

    def connz(
        self,
        sort: t.Union[str, SortOption] = SortOption.CID,
        auth: bool = False,
        subs: t.Union[bool, str, SubsOption] = SubsOption.FALSE,
        offset: int = 0,
        limit: int = 1024,
        cid: t.Optional[int] = None,
        state: t.Union[str, StateOption] = StateOption.OPEN,
        mqtt_client: t.Optional[str] = None,
    ) -> t.Dict[str, t.Any]:
        """The /connz endpoint reports more detailed information on current and recently closed connections.

        It uses a paging mechanism which defaults to 1024 connections.

        Arguments:
            sort: sorts the results. Default is connection ID.
            auth: include username. Default is False.
            subs: include subscriptions. Default is False. When set to "detail", a list with more detailed subscription information is returned.
            offset: pagination offset. Default is 0.
            limit: number of results to return. Default is 1024.
            cid: return result for a single connection by its id. Omitted by default.
            state: return results for connections of particular state. Default is "open".
            mqtt_client: return results for connections with this MQTT client id. Omitted by default.

        Returns:
            results as a dictionary.
        """
        if not isinstance(sort, SortOption):
            sort = SortOption(sort)
        if not isinstance(subs, SubsOption):
            subs = SubsOption(subs)
        if not isinstance(state, StateOption):
            state = StateOption(state)
        params = {
            "sort": sort.value,
            "auth": 1 if auth else 0,
            "subs": subs.value,
            "offset": offset,
            "limit": limit,
        }
        if cid:
            params["cid"] = int(cid)
        if mqtt_client:
            params["mqtt_client"] = mqtt_client
        return self._request("/connz", **params)

    def accountz(self, acc: t.Optional[str] = None) -> t.Dict[str, t.Any]:
        """The /accountz endpoint reports information on a server's active accounts.

        The default behavior is to return a list of all accounts known to the server.

        Arguments:
            acc: include metrics for the specified account only. Default is empty. When not set
                a list of all accounts is included.
        """
        params: t.Dict[str, t.Any] = {}
        if acc:
            params["acc"] = acc
        return self._request("/accountz", **params)

    def accstatz(self, unused: bool = False) -> t.Dict[str, t.Any]:
        """The /accstatz endpoint reports per-account statistics such as the number of connections, messages/bytes in/out, etc.

        Arguments:
            unused: include accounts that do not have any current connections when True. Default is False.

        Returns:
            results as a dictionary.
        """
        return self._request("/accstatz", unused=unused)

    def subsz(
        self,
        subs: bool = False,
        offset: int = 0,
        limit: int = 1024,
        test: t.Optional[str] = None,
    ) -> t.Dict[str, t.Any]:
        """The /subsz endpoint reports detailed information about the current subscriptions and the routing data structure.
        It is not normally used.

        Arguments:
            subs: include subscriptions. Default is false.
            offset: pagination offset. Default is 0.
            limit: number of results to return. Default is 1024.
            test: test whether a subscription exists.

        Returns:
            results as a dictionary.
        """
        params: t.Dict[str, t.Any] = {
            "subs": subs,
            "offset": offset,
            "limit": limit,
        }
        if test:
            params["test"] = test
        return self._request("/subsz", **params)

    def routez(
        self, subs: t.Union[bool, str, SubsOption] = SubsOption.FALSE
    ) -> t.Dict[str, t.Any]:
        """The /routez endpoint reports information on active routes for a cluster.

        Routes are expected to be low, so there is no paging mechanism with this endpoint.

        Arguments:
            subs: include subscriptions. Default is False. When set to "detail", a list with more details subscription information is returned.

        Returns:
            results as a dictionary.
        """
        if not isinstance(subs, SubsOption):
            subs = SubsOption(subs)
        return self._request("/routez", subs=subs.value)

    def leafz(self, subs: bool = False) -> t.Dict[str, t.Any]:
        """The /leafz endpoint reports detailed information about the leaf node connections.

        Arguments:
            subs: include internal subscriptions. Default is False.

        Returns:
            results as dict
        """
        return self._request("/leafz", subs=subs)

    def gatewayz(
        self,
        accs: bool = False,
        gw_name: t.Optional[str] = None,
        acc_name: t.Optional[str] = None,
    ) -> t.Dict[str, t.Any]:
        """The /gatewayz endpoint reports information about gateways used to create a NATS supercluster.

        Like routes, the number of gateways are expected to be low, so there is no paging mechanism with this endpoint.

        Arguments:
            accs: include account information. Default is false.
            gw_name: return only remote gateways with this name. Omitted by default.
            acc_name: limit the list of accounts to this account name. Omitted by default.

        Returns:
            results as dict
        """
        params: t.Dict[str, t.Any] = {"accs": bool(accs)}
        if gw_name:
            params["gw_name"] = gw_name
        if acc_name:
            params["acc_name"] = acc_name
        return self._request("/gatewayz", **params)

    def healthz(
        self,
        js_enabled: bool = False,
        js_server_only: bool = False,
    ) -> t.Dict[str, t.Any]:
        """The /healthz endpoint returns OK if the server is able to accept connections.

        Arguments:
            js_enabled: returns an error if jetstream is disabled. Omitted by default.
            js_server_only: skip healthcheck of accounts, streams and consumers. Omitted by default.

        Returns:
            results as dictionary.
        """
        params: t.Dict[str, t.Any] = {}
        if js_enabled:
            params["js_enabled"] = 1
        if js_server_only:
            params["js_server_only"] = 1
        return self._request("/healthz", **params)

accountz(acc=None)

The /accountz endpoint reports information on a server's active accounts.

The default behavior is to return a list of all accounts known to the server.

Parameters:

Name Type Description Default
acc t.Optional[str]

include metrics for the specified account only. Default is empty. When not set a list of all accounts is included.

 None
Source code in nats_tools/monitor.py 
163
164
165
166
167
168
169
170
171
172
173
174
175
def accountz(self, acc: t.Optional[str] = None) -> t.Dict[str, t.Any]:
    """The /accountz endpoint reports information on a server's active accounts.

    The default behavior is to return a list of all accounts known to the server.

    Arguments:
        acc: include metrics for the specified account only. Default is empty. When not set
            a list of all accounts is included.
    """
    params: t.Dict[str, t.Any] = {}
    if acc:
        params["acc"] = acc
    return self._request("/accountz", **params)

accstatz(unused=False)

The /accstatz endpoint reports per-account statistics such as the number of connections, messages/bytes in/out, etc.

Parameters:

Name Type Description Default
unused bool

include accounts that do not have any current connections when True. Default is False.

 False

Returns:

Type Description
t.Dict[str, t.Any]

results as a dictionary.
Source code in nats_tools/monitor.py 
177
178
179
180
181
182
183
184
185
186
def accstatz(self, unused: bool = False) -> t.Dict[str, t.Any]:
    """The /accstatz endpoint reports per-account statistics such as the number of connections, messages/bytes in/out, etc.

    Arguments:
        unused: include accounts that do not have any current connections when True. Default is False.

    Returns:
        results as a dictionary.
    """
    return self._request("/accstatz", unused=unused)

connz(sort=SortOption.CID, auth=False, subs=SubsOption.FALSE, offset=0, limit=1024, cid=None, state=StateOption.OPEN, mqtt_client=None)

The /connz endpoint reports more detailed information on current and recently closed connections.

It uses a paging mechanism which defaults to 1024 connections.

Parameters:

Name Type Description Default
sort t.Union[str, SortOption]

sorts the results. Default is connection ID.

 SortOption.CID
auth bool

include username. Default is False.

 False
subs t.Union[bool, str, SubsOption]

include subscriptions. Default is False. When set to "detail", a list with more detailed subscription information is returned.

 SubsOption.FALSE
offset int

pagination offset. Default is 0.

 0
limit int

number of results to return. Default is 1024.

 1024
cid t.Optional[int]

return result for a single connection by its id. Omitted by default.

 None
state t.Union[str, StateOption]

return results for connections of particular state. Default is "open".

 StateOption.OPEN
mqtt_client t.Optional[str]

return results for connections with this MQTT client id. Omitted by default.

 None

Returns:

Type Description
t.Dict[str, t.Any]

results as a dictionary.
Source code in nats_tools/monitor.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def connz(
    self,
    sort: t.Union[str, SortOption] = SortOption.CID,
    auth: bool = False,
    subs: t.Union[bool, str, SubsOption] = SubsOption.FALSE,
    offset: int = 0,
    limit: int = 1024,
    cid: t.Optional[int] = None,
    state: t.Union[str, StateOption] = StateOption.OPEN,
    mqtt_client: t.Optional[str] = None,
) -> t.Dict[str, t.Any]:
    """The /connz endpoint reports more detailed information on current and recently closed connections.

    It uses a paging mechanism which defaults to 1024 connections.

    Arguments:
        sort: sorts the results. Default is connection ID.
        auth: include username. Default is False.
        subs: include subscriptions. Default is False. When set to "detail", a list with more detailed subscription information is returned.
        offset: pagination offset. Default is 0.
        limit: number of results to return. Default is 1024.
        cid: return result for a single connection by its id. Omitted by default.
        state: return results for connections of particular state. Default is "open".
        mqtt_client: return results for connections with this MQTT client id. Omitted by default.

    Returns:
        results as a dictionary.
    """
    if not isinstance(sort, SortOption):
        sort = SortOption(sort)
    if not isinstance(subs, SubsOption):
        subs = SubsOption(subs)
    if not isinstance(state, StateOption):
        state = StateOption(state)
    params = {
        "sort": sort.value,
        "auth": 1 if auth else 0,
        "subs": subs.value,
        "offset": offset,
        "limit": limit,
    }
    if cid:
        params["cid"] = int(cid)
    if mqtt_client:
        params["mqtt_client"] = mqtt_client
    return self._request("/connz", **params)

gatewayz(accs=False, gw_name=None, acc_name=None)

The /gatewayz endpoint reports information about gateways used to create a NATS supercluster.

Like routes, the number of gateways are expected to be low, so there is no paging mechanism with this endpoint.

Parameters:

Name Type Description Default
accs bool

include account information. Default is false.

 False
gw_name t.Optional[str]

return only remote gateways with this name. Omitted by default.

 None
acc_name t.Optional[str]

limit the list of accounts to this account name. Omitted by default.

 None

Returns:

Type Description
t.Dict[str, t.Any]

results as dict
Source code in nats_tools/monitor.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
def gatewayz(
    self,
    accs: bool = False,
    gw_name: t.Optional[str] = None,
    acc_name: t.Optional[str] = None,
) -> t.Dict[str, t.Any]:
    """The /gatewayz endpoint reports information about gateways used to create a NATS supercluster.

    Like routes, the number of gateways are expected to be low, so there is no paging mechanism with this endpoint.

    Arguments:
        accs: include account information. Default is false.
        gw_name: return only remote gateways with this name. Omitted by default.
        acc_name: limit the list of accounts to this account name. Omitted by default.

    Returns:
        results as dict
    """
    params: t.Dict[str, t.Any] = {"accs": bool(accs)}
    if gw_name:
        params["gw_name"] = gw_name
    if acc_name:
        params["acc_name"] = acc_name
    return self._request("/gatewayz", **params)

healthz(js_enabled=False, js_server_only=False)

The /healthz endpoint returns OK if the server is able to accept connections.

Parameters:

Name Type Description Default
js_enabled bool

returns an error if jetstream is disabled. Omitted by default.

 False
js_server_only bool

skip healthcheck of accounts, streams and consumers. Omitted by default.

 False

Returns:

Type Description
t.Dict[str, t.Any]

results as dictionary.
Source code in nats_tools/monitor.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def healthz(
    self,
    js_enabled: bool = False,
    js_server_only: bool = False,
) -> t.Dict[str, t.Any]:
    """The /healthz endpoint returns OK if the server is able to accept connections.

    Arguments:
        js_enabled: returns an error if jetstream is disabled. Omitted by default.
        js_server_only: skip healthcheck of accounts, streams and consumers. Omitted by default.

    Returns:
        results as dictionary.
    """
    params: t.Dict[str, t.Any] = {}
    if js_enabled:
        params["js_enabled"] = 1
    if js_server_only:
        params["js_server_only"] = 1
    return self._request("/healthz", **params)

jsz(acc=None, accounts=None, streams=None, consumers=None, config=None, leader_only=False, offset=0, limit=1024)

The /jsz endpoint reports more detailed information on JetStream.

For accounts, it uses a paging mechanism that defaults to 1024 connections.

If you're in a clustered environment, it is recommended to retrieve the information

from the stream's leader in order to get the most accurate and up-to-date data.

Parameters:

Name Type Description Default
acc t.Optional[str]

include metrics for the specified account only. Omitted by default.

 None
accounts t.Optional[bool]

include account specific jetstream information. Default is False.

 None
streams t.Optional[bool]

include streams. When set, implies accounts=True. Default is False.

 None
consumers t.Optional[bool]

include consumers. When set, implies stream=True. Default is False.

 None
config t.Optional[bool]

when stream or consumer are requested, include their respective configuration. Default is False.

 None
leader_only bool

only the leader responds. Default is False.

 False
offset int

pagination offset. Default is 0.

 0
limit int

number of results to return. Default is 1024.

 1024

Returns:

Type Description
t.Dict[str, t.Any]

results as a dictionary.
Source code in nats_tools/monitor.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def jsz(
    self,
    acc: t.Optional[str] = None,
    accounts: t.Optional[bool] = None,
    streams: t.Optional[bool] = None,
    consumers: t.Optional[bool] = None,
    config: t.Optional[bool] = None,
    leader_only: bool = False,
    offset: int = 0,
    limit: int = 1024,
) -> t.Dict[str, t.Any]:
    """The /jsz endpoint reports more detailed information on JetStream.

    For accounts, it uses a paging mechanism that defaults to 1024 connections.

    NOTE: If you're in a clustered environment, it is recommended to retrieve the information
          from the stream's leader in order to get the most accurate and up-to-date data.

    Arguments:
        acc: include metrics for the specified account only. Omitted by default.
        accounts: include account specific jetstream information. Default is False.
        streams: include streams. When set, implies `accounts=True`. Default is False.
        consumers: include consumers. When set, implies `stream=True`. Default is False.
        config: when stream or consumer are requested, include their respective configuration. Default is False.
        leader_only: only the leader responds. Default is False.
        offset: pagination offset. Default is 0.
        limit: number of results to return. Default is 1024.

    Returns:
        results as a dictionary.
    """
    params: t.Dict[str, t.Any] = {
        "leader-only": leader_only,
        "limit": limit,
        "offset": offset,
    }
    if accounts:
        params["accounts"] = accounts
    if streams:
        params["streams"] = streams
    if consumers:
        params["consumers"] = consumers
    if config:
        params["config"] = config
    if acc:
        params["acc"] = acc
    return self._request("/jsz", **params)

leafz(subs=False)

The /leafz endpoint reports detailed information about the leaf node connections.

Parameters:

Name Type Description Default
subs bool

include internal subscriptions. Default is False.

 False

Returns:

Type Description
t.Dict[str, t.Any]

results as dict
Source code in nats_tools/monitor.py 
233
234
235
236
237
238
239
240
241
242
def leafz(self, subs: bool = False) -> t.Dict[str, t.Any]:
    """The /leafz endpoint reports detailed information about the leaf node connections.

    Arguments:
        subs: include internal subscriptions. Default is False.

    Returns:
        results as dict
    """
    return self._request("/leafz", subs=subs)

routez(subs=SubsOption.FALSE)

The /routez endpoint reports information on active routes for a cluster.

Routes are expected to be low, so there is no paging mechanism with this endpoint.

Parameters:

Name Type Description Default
subs t.Union[bool, str, SubsOption]

include subscriptions. Default is False. When set to "detail", a list with more details subscription information is returned.

 SubsOption.FALSE

Returns:

Type Description
t.Dict[str, t.Any]

results as a dictionary.
Source code in nats_tools/monitor.py 
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def routez(
    self, subs: t.Union[bool, str, SubsOption] = SubsOption.FALSE
) -> t.Dict[str, t.Any]:
    """The /routez endpoint reports information on active routes for a cluster.

    Routes are expected to be low, so there is no paging mechanism with this endpoint.

    Arguments:
        subs: include subscriptions. Default is False. When set to "detail", a list with more details subscription information is returned.

    Returns:
        results as a dictionary.
    """
    if not isinstance(subs, SubsOption):
        subs = SubsOption(subs)
    return self._request("/routez", subs=subs.value)

subsz(subs=False, offset=0, limit=1024, test=None)

The /subsz endpoint reports detailed information about the current subscriptions and the routing data structure. It is not normally used.

Parameters:

Name Type Description Default
subs bool

include subscriptions. Default is false.

 False
offset int

pagination offset. Default is 0.

 0
limit int

number of results to return. Default is 1024.

 1024
test t.Optional[str]

test whether a subscription exists.

 None

Returns:

Type Description
t.Dict[str, t.Any]

results as a dictionary.
Source code in nats_tools/monitor.py 
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def subsz(
    self,
    subs: bool = False,
    offset: int = 0,
    limit: int = 1024,
    test: t.Optional[str] = None,
) -> t.Dict[str, t.Any]:
    """The /subsz endpoint reports detailed information about the current subscriptions and the routing data structure.
    It is not normally used.

    Arguments:
        subs: include subscriptions. Default is false.
        offset: pagination offset. Default is 0.
        limit: number of results to return. Default is 1024.
        test: test whether a subscription exists.

    Returns:
        results as a dictionary.
    """
    params: t.Dict[str, t.Any] = {
        "subs": subs,
        "offset": offset,
        "limit": limit,
    }
    if test:
        params["test"] = test
    return self._request("/subsz", **params)

varz()

The /varz endpoint returns general information about the server state and configuration.

Example: https://demo.nats.io:8222/varz

Source code in nats_tools/monitor.py 
61
62
63
64
65
66
def varz(self) -> t.Dict[str, t.Any]:
    """The /varz endpoint returns general information about the server state and configuration.

    Example: https://demo.nats.io:8222/varz
    """
    return self._request("/varz")