sbusr 文档¶

executor module¶

该模块定义了 RPC 任务执行器

服务器模块 server 将来自 smartbus 的请求传递给执行器类 Executor 的实例， 执行器解析请求的 JSON RPC 格式，并分配执行体，在线程池中执行， 然后返回结果。

class executor.Executor(queue_maxsize=0, pool_processes=None, pool_maxtasksperchild=None)¶

基类：logging.handlers.QueueListener

smarbus JSON RPC 请求执行器

使用进程池执行 RPC 请求

参数:	queue_maxsize – 任务队列最大值 默认为0，表示无限制。 pool_processes – 执行器池的最大数量 默认为 none，表示使用 CPU 核心数量作为其最大值 pool_maxtasksperchild – 进程池最大执行数量 默认为 None，表示无限制。超过该值，则重启子进程。仅对进程池模型有效。

在接收到 smartbus 请求后，需要向这个队列放置数据，数据的格式是： client, pack_info, txt 分别是：收到数据的 smartbus 客户端的实例，数据包附加信息，数据文本。

收到数据后，本类型的实例将按照 JSON-RPC 格式解析数据，并执行 JSON RPC 请求，最后将执行结果通过 smartbus 客户端进行返回。 返回数据格式是符合 JSON RPC 标准的字符串。

handle(record)¶

put(client, pack_info, txt)¶

start()¶

stop()¶

globalvars module¶

记录一些全局变量，特别是需要跨进程访问的全局变量

globalvars.executor = None¶

全局 RPC 执行器

类型是 executor.Executor

注意

仅在主进程有效

globalvars.ipc_smartbusclient = None¶

IPC客户端

注意

仅在主进程有效

globalvars.ipsc_set = set()¶

IPSC节点集合

一个 set 类型全局变量 进程池模式中，它仅在主进程有效

将Smartbus上的IPSC客户端记录在这个集合中，其中每个元素表示一个IPSC客户端连接，其格式是:

(unitid, clientid)

注意

仅在主进程有效

globalvars.main_logging_listener = None¶

多进程的全局日志监听器

注意

仅在主进程有效

globalvars.main_logging_queue = None¶

多进程的全局日志队列

注意

仅在主进程有效

globalvars.net_smartbusclients = []¶

NET客户端列表

注意

仅在主进程有效

globalvars.prog_args = None¶

命令行参数

注解

在主进程初始化进程池的时候，该变量在子进程中被赋值

jsonrpc module¶

JSON-RPC 常用方法

exception jsonrpc.Error(id_=None, code=None, message=None, data=None)¶

基类：builtins.Exception

json rpc 的 Error Response 对象

to_dict()¶

转为 dict

to_json()¶

转为 JSON 字符串

exception jsonrpc.FormatError(id_=None, code=-32700, message='Invalid JSON was received by the server. An error occurred on the server while parsing the JSON text.', data=None)¶

基类：jsonrpc.Error

JSON-RPC格式错误

exception jsonrpc.InvalidErrorError(id_=None, code=-32600, message='The JSON sent is not a valid Error object.', data=None)¶

基类：jsonrpc.Error

无效的JSON-RPC错误回复

exception jsonrpc.InvalidParamsError(id_=None, code=-32602, message='Invalid method parameter(s).', data=None)¶

基类：jsonrpc.Error

无效的JSON-RPC请求参数

exception jsonrpc.InvalidRequestError(id_=None, code=-32600, message='The JSON sent is not a valid Request object.', data=None)¶

基类：jsonrpc.Error

无效的JSON-RPC请求

exception jsonrpc.InvalidResponseError(id_=None, code=-32600, message='The JSON sent is not a valid Response object.', data=None)¶

基类：jsonrpc.Error

无效的JSON-RPC回复

exception jsonrpc.MethodNotFoundError(id_=None, code=-32601, message='The method does not exist / is not available.', data=None)¶

基类：jsonrpc.Error

JSON-RPC请求方法未找到

exception jsonrpc.RpcTimeoutError¶

基类：builtins.Exception

jsonrpc.parse(txt)¶

解析 JSON RPC

参数:	txt – 带解析的 JSON 字符串
返回:	request, result, error 如果是request，第一个参数返回该请求的dict，格式是： {'id': 'id-of-the-request', 'method': 'your_method', params['what', 'ever', 'params']} 其它两个参数是 None 如果是返回结果，第二个参数返回结果 dict，格式是： {'id': 'id-of-the-request', 'result': 'your_result'} 其它两个参数是 None

jsonrpc.recursive_jsonable(obj, encoding='utf-8')¶

递归转为可JSON序列化数据类型

参数:	obj – 要转化的对象，可以是基本数据类型，或者 dict, list, tuple, date, datetime encoding – 编码，默认是utf8
返回:	转换后对象

jsonrpc.to_bytes(s, encoding='utf-8')¶

jsonrpc.to_str(s, encoding='utf-8')¶

jsonrpc.to_unicode(s, encoding='utf-8')¶

loggingqueue module¶

class loggingqueue.QueueHandler(queue)¶

基类：logging.Handler

This handler sends events to a queue. Typically, it would be used together with a multiprocessing Queue to centralise logging to file in one process (in a multi-process application), so as to avoid file write contention between processes.

This code is new in Python 3.2, but this class can be copy pasted into user code for use with earlier Python versions.

emit(record)¶

Emit a record.

Writes the LogRecord to the queue, preparing it for pickling first.

enqueue(record)¶

Enqueue a record.

The base implementation uses put_nowait. You may want to override this method if you want to use blocking, timeouts or custom queue implementations.

prepare(record)¶

Prepares a record for queuing. The object returned by this method is enqueued.

The base implementation formats the record to merge the message and arguments, and removes unpickleable items from the record in-place.

You might want to override this method if you want to convert the record to a dict or JSON string, or send a modified copy of the record while leaving the original intact.

class loggingqueue.QueueListener(queue, *handlers)¶

基类：builtins.object

This class implements an internal threaded listener which watches for LogRecords being added to a queue, removes them and passes them to a list of handlers for processing.

dequeue(block)¶

Dequeue a record and return it, optionally blocking.

The base implementation uses get. You may want to override this method if you want to use timeouts or work with custom queue implementations.

enqueue_sentinel()¶

This is used to enqueue the sentinel record.

The base implementation uses put_nowait. You may want to override this method if you want to use timeouts or work with custom queue implementations.

handle(record)¶

Handle a record.

This just loops through the handlers offering them the record to handle.

prepare(record)¶

Prepare a record for handling.

This method just returns the passed-in record. You may want to override this method if you need to do any custom marshalling or manipulation of the record before passing it to the handlers.

start()¶

Start the listener.

This starts up a background thread to monitor the queue for LogRecords to process.

stop()¶

Stop the listener.

This asks the thread to terminate, and then waits for it to do so. Note that if you don’t call this before your application exits, there may be some records still left on the queue, which won’t be processed.

methods package¶

sbusr_run module¶

sbsur_run 命令行

运行 sbusr 服务程序

在当前终端/命令行窗口中运行该程序

直接运行：

POSIX（需要权限）:

./sbusr_run.py <option>

Windows（需要权限与文件关联）:

sbusr_run.py <option>        

使用 Python 运行:

python sbusr_run.py <option>

-V, --version¶

显示版本信息。

-v, --verbose¶

输出更详细的日志。默认为 False 。

-h, --help¶

输出帮助信息

-W, --no-web-server¶

不启动该程序的内嵌web服务器模块

警告

如果不启动Web服务器模块，则无法调用流程，也无法通过命令重启子进程。

server module¶

服务

该模块负责:

• 维持smartbus客户端的连接
• 分配及执行来自smartbus的任务，并返回结果。
• 启动Web程序

server.run(args)¶

运行服务

参数:	args – 命令行参数

注意

该函数将“阻塞”，直到服务结束才会返回

server.stop()¶

停止服务

注意

该函数是异步的，在它执行后， server.run() 将会退出 IOLoop ，但是 stop 的返回与之无关。

settings module¶

配置

settings.EXECUTOR_CONFIG = {'pool_processes': 1, 'pool_maxtasksperchild': 1000, 'queue_maxsize': 1000}¶

执行器设置

这个设置被用于 executor.Executor 构造函数的传入参数。

参数:	queue_maxsize – 任务队列最大值。0，表示无限制。 pool_processes – 执行器池的最大数量。 None 表示使用 CPU 核心数量作为其最大值。 pool_maxtasksperchild – 进程池最大执行数量。 None 表示无限制。超过该值，则重启子进程。

警告

不得删除该变量，不得修改该变量的结构。

settings.FLOW_ACK_TIMEOUT = 15¶

流程调用时等待调用结果的最大时间，单位是秒

settings.LOGGING_CONFIG = {'loggers': {'smartbus.ipcclient.client.Client': {'level': 'INFO', 'handlers': ['smartbusFile']}, 'smartbus.netclient.client.Client': {'level': 'INFO', 'handlers': ['smartbusFile']}}, 'version': 1, 'root': {'level': 'INFO', 'handlers': ['console', 'file', 'fileErr']}, 'handlers': {'file': {'formatter': 'normal', 'backupCount': 1000, 'level': 'DEBUG', 'maxBytes': 10485760, 'filename': '../logs/log', 'class': 'logging.handlers.RotatingFileHandler'}, 'consoleErr': {'formatter': 'normal', 'stream': 'ext://sys.stderr', 'level': 'ERROR', 'class': 'logging.StreamHandler'}, 'smartbusFile': {'formatter': 'normal', 'filename': '../logs/smartbus', 'backupCount': 1000, 'maxBytes': 10485760, 'class': 'logging.handlers.RotatingFileHandler'}, 'fileErr': {'formatter': 'normal', 'backupCount': 1000, 'level': 'ERROR', 'maxBytes': 10485760, 'filename': '../logs/err', 'class': 'logging.handlers.RotatingFileHandler'}, 'console': {'formatter': 'normal', 'stream': 'ext://sys.stdout', 'class': 'logging.StreamHandler'}}, 'formatters': {'normal': {'format': '%(asctime)s <%(processName)-10s,%(threadName)-10s> %(levelname)-8s %(name)s - %(message)s'}}}¶

日志设置

其格式请参考: https://docs.python.org/2/library/logging.config.html#configuration-dictionary-schema 与 https://docs.python.org/3/library/logging.config.html#configuration-dictionary-schema

settings.SMARTBUS_CONFIG = {'type': 'net', 'initialize': {'unitid': 19}, 'clients': [{'masterPort': 8089, 'encoding': 'utf-8', 'localClientId': 0, 'authorPwd': None, 'authorUsr': None, 'slaverHost': None, 'extInfo': None, 'localClientType': 19, 'slaverPort': 0, 'masterHost': '10.4.62.45'}]}¶

SmartBus客户端配置

参数:

type – smartbus客户端类型，有效值："net" 或 "ipc" "net" : 使用smartbus网络通信客户端 "ipc" : 使用smartbus进程通信客户端 initialize – smartbus客户端模块初始化参数属性字段 当 type 属性为 "net" 时，该属性必须设置一个属性 unitid 当 type 属性为 "ipc" 时，该属性必须设置两个属性 clientid 与 clienttype 具体含义请参考 smartbus　文档 instance – smartbus进程通信客户端地址参数属性字段 该 dict 属性需要设置3个属性： username password extInfo 具体含义请参考 smartbus　文档 注意 仅当 type 属性为 "ipc" 时有效 clients – smartbus网络通信客户端地址参数属性字段 该 list 属性的每个成员表示一个smartbus网络通信客户端，每个客户端的参数都是： localClientId localClientType masterHost masterPort slaverHost slaverPort authorUsr authorPwd extInfo 具体含义请参考 smartbus　文档 注意 仅当 type 属性为 "net" 时有效

settings.SMARTBUS_NOTIFY_TTL = 10000¶

通过smartbus发送给IPSC流程的notify消息的生存期（ms）

settings.WEBSERVER_LISTEN = (8080, '')¶

WEB 服务器监听端口与地址.

格式是:

(port, address)

警告

不得删除该变量，不得修改该变量的结构。

webhandlers module¶

Web Request Handlers

class webhandlers.FlowHandler(application, request, **kwargs)¶

基类：tornado.web.RequestHandler

POST启动流程

on_connection_close()¶

post(*args, **kwargs)¶

classmethod set_flow_ack(ack)¶

设置流程启动执行回执

参数:	ack (webhandlers.FlowInvokeAck) – 回执信息

class webhandlers.FlowInvokeAck(packinfo, project, invokeid, ack, msg)¶

基类：builtins.object

class webhandlers.ResetHandler(application, request, **kwargs)¶

基类：tornado.web.RequestHandler

get()¶

reset_executor()¶