烧瓶的简单avro ipc端点注册
flask-avro的Python项目详细描述
#flask avro
这是一个简单的flask avro ipc端点注册扩展。
有关avro签出的信息：
*[https://avro.apache.org/]（https://avro.apache.org/）
*[http://avro.apache.org/docs/current/spec.html]（http://avro.apache.org/docs/current/spec.html）
向特定的url路由注册avro协议。这允许您处理协议中定义的消息。
“name”：“from”，
“type”：“string”
}，
{
“name”：“body”，
“type”：“string”
}
]
}
}
}
]，
“消息”：{
“回音消息”：{
“请求”：[
{
“名称”：“消息”，
“类型”：“消息”
}
]，
“响应”：“字符串”，
“错误”：[]
}
}
}
`````
实现消息“echo_message”逻辑的简单flask应用程序示例，您可以执行以下操作：
````
from flask import flask
from flask.ext.avro import flask avro endpoint
@test_avro_endpoint.register（“echo_message”）：
defhandle_test_send（request_message）：
return request_message.get（“body”）
test_u avro_u endpoint.connect（app，“/”
`````
以上代码注册了一个函数，当端点接收到“echo_x message”消息时，该函数将处理请求消息。函数还负责返回作为协议响应所需的正确对象。
您必须安装带有pip的“avro”包，然后您的客户端可以如下所示：
`````
from avro.protocol import protocol
from avro.ipc import requestor，httpreceptory
client=httpreceptory（“localhost”，9090）
请求者=请求者（测试协议，客户机）
测试avro端点的测试助手。通常，在测试视图时，需要使用“test_flask_app.test_client（）”生成对视图终结点的请求。但是，创建avro请求消息有点麻烦。
为了方便测试，有一个测试实用程序类“flask.ext.avro.util.flaskavrotestclient”。这是测试客户机的一个方便包装器。
示例用法：
````
来自unittest import testcase
来自flask.ext.avro.util import flaskavrotestclient
来自avro.protocol import protocol
test protocol=protocol.parse（open（“test.avpr”）.read（））
class testavroendpoint（testcase）：
def setup（self）：
生成测试应用程序或提取已创建的应用程序
…
self.app=test应用程序
def test应用程序终结点（self）：
client=flaskavrotestclient（test_protocol，self.app.test_client（），“/”
test_body=”服用蛋白质药丸并将其放入lmet on“
test_message=test_message={
”from“：”地面控制“，
”body“：test_body
}
response=client.request（“echo_message”，test_message）
self.assertequal（test_body，响应）
`````
-----------------------------------
0.0.1
----------
-初始公共版本。-
这是一个简单的flask avro ipc端点注册扩展。
有关avro签出的信息：
*[https://avro.apache.org/]（https://avro.apache.org/）
*[http://avro.apache.org/docs/current/spec.html]（http://avro.apache.org/docs/current/spec.html）
向特定的url路由注册avro协议。这允许您处理协议中定义的消息。
“name”：“from”，
“type”：“string”
}，
{
“name”：“body”，
“type”：“string”
}
]
}
}
}
]，
“消息”：{
“回音消息”：{
“请求”：[
{
“名称”：“消息”，
“类型”：“消息”
}
]，
“响应”：“字符串”，
“错误”：[]
}
}
}
`````
实现消息“echo_message”逻辑的简单flask应用程序示例，您可以执行以下操作：
````
from flask import flask
from flask.ext.avro import flask avro endpoint
@test_avro_endpoint.register（“echo_message”）：
defhandle_test_send（request_message）：
return request_message.get（“body”）
test_u avro_u endpoint.connect（app，“/”
`````
以上代码注册了一个函数，当端点接收到“echo_x message”消息时，该函数将处理请求消息。函数还负责返回作为协议响应所需的正确对象。
您必须安装带有pip的“avro”包，然后您的客户端可以如下所示：
`````
from avro.protocol import protocol
from avro.ipc import requestor，httpreceptory
请求者=请求者（测试协议，客户机）
测试avro端点的测试助手。通常，在测试视图时，需要使用“test_flask_app.test_client（）”生成对视图终结点的请求。但是，创建avro请求消息有点麻烦。
为了方便测试，有一个测试实用程序类“flask.ext.avro.util.flaskavrotestclient”。这是测试客户机的一个方便包装器。
示例用法：
````
来自unittest import testcase
来自flask.ext.avro.util import flaskavrotestclient
来自avro.protocol import protocol
test protocol=protocol.parse（open（“test.avpr”）.read（））
class testavroendpoint（testcase）：
def setup（self）：
生成测试应用程序或提取已创建的应用程序
…
self.app=test应用程序
def test应用程序终结点（self）：
client=flaskavrotestclient（test_protocol，self.app.test_client（），“/”
test_body=”服用蛋白质药丸并将其放入lmet on“
test_message=test_message={
”from“：”地面控制“，
”body“：test_body
}
response=client.request（“echo_message”，test_message）
self.assertequal（test_body，响应）
`````
-----------------------------------
0.0.1
----------
-初始公共版本。-
欢迎加入QQ群-->： 979659372
