salesforce零基础学习（八十五）streaming api 简单使用（接近实时获取你需要跟踪的数据的更新消息状态）

Posted 2020-10-22 zero.zhang

Streaming API参考链接：

https://trailhead.salesforce.com/en/modules/api_basics/units/api_basics_streaming

 https://resources.docs.salesforce.com/210/latest/en-us/sfdc/pdf/api_streaming.pdf

背景：工作中我们有可能会有这样相关的需求：某些数据很重要，需要实时监控是否有变化,或者某些数据在其他的平台有集成。如果有变化，不刷新页面或者做其他trigger等操作便可以收到相关更改的推送通知（不管是salesforce平台还是其他外部服务，比如java端等，可以通过此种方式进行实时同步），告诉当前的需要监控的记录已经发生了变化。这种场景往往更会出现在数据不止存储在salesforce端，还有其他平台有数据的存储或者访问。这种情况下可以使用Streaming API。

一.Stream API简单介绍

Streaming API提供了两种功能，一种为基于Salesforce数据改变，对订阅的客户端进行通知推送，另一种是不基于Salesforce数据改变，对订阅的客户端进行通知推送，接下来大部分内容基于Salesforce数据改变后，对订阅的客户端进行推送，另外一种感兴趣的可以自行查看。通知推送不止可以应用于salesforce系统，其他系统等也可以通过oauth等连接后接收到通知推送，感兴趣的可以查看API提供的内容。

Streaming API用于制定一套你想要接收的数据的条件，并且指定哪些事件（create/update/delete/undelete）情况下符合的数据推送一条通知到客户端，一条通知就是事件触发的结果发送到渠道的消息。通知的格式为JSON的格式。使用Streaming API可以减少不必要的API请求。

Streaming API使用的推送技术，即服务器端会主动给订阅的客户端发送通知信息，而不是客户端去调用服务器端返回消息，使用Bayeux协议和CometD用于长轮询。

cometd 参考链接： https://docs.cometd.org/

如果想使用Streaming API，需要enable api的权限以及streaming api的权限

想要接受通知(notifications),当前登录的user必须针对StreamingChannel表拥有read权限

想要创建和管理通知，当前登录的user必须对StreamingChannel拥有Create权限

 

 二.Streaming API 使用步骤

针对开发者来说，更关注的是这个东西如何使用。使用Streaming API实现订阅者接收符合条件的推送消息可以三步走：

1.创建PushTopic

去除一下PushTopic神奇的面纱，简单的来说，PushTopic是一个标准的sObject，封装了以下的字段：

• ApiVersion：用于指定查询的SQL的版本，37以后系统可以存储24小时以内的事件，必填字段；
• Description：PushTopic的描述信息，限定在400个字符以内；
• ID：指定一条记录的全局唯一的标识；
• isActive：是否可用，关系到PushTopic的限制计数（系统对PushTopic有limitation size）；
• IsDeleted：指定此PushTopic是否移动到回收站；
• Name：PushTopic的名字，定义了渠道的名字，并且此名称必须是唯一的，后期订阅者订阅时，使用的就是这个名字；
• NotifyForFields：指定哪些字段被评估生成通知；
• NotifyForOperations：指定数据哪种事件操作会生成通知，在api version 29以后，此字段为只读字段；
• NotifyForOperationCreate：Create操作是否会生成通知，api29以后可用；
• NotifyForOperationDelete：Delete操作是否会生成通知，api29以后可用；
• NotifyForOperationUndelete：UnDelete操作是否会生成通知，api29以后可用；
• NotifyForOperationUpdate：Update操作是否会生成通知，api29以后可用；
• Query：SOQL语句决定了哪些数据符合触发的事件后会发送到渠道。

恩，新建一个PushTopic，设置了必填字段的值insert以后，一个PushTopic就创建完了。这些字段有几个需要详细的描述一下：

Query：Query在PushTopic的作用不言而喻，定义了哪些数据可以满足条件进行推送。Query语法和SOQL基本相同，但是有一些情况不支持。限制如下：

• select字段无ID；
• 子连接查询Semi-joins and anti-joins；
• 聚合类的查询，比如SUM，AVG等；
• COUNT；
• LIMIT；
• 关系型查询不支持，只可以查询到ID，不能通过“.”的方式查询到父的其他信息；
• 查询中包括Text Area字段；
• ORDER BY；
• GROUP BY；
• WHERE部分使用了formula类型字段；
• NOT   Example : SELECT Id FROM Account WHERE NOT Name = \'Salesforce.com\'
• OFFSET
• TYPEOF

NotifyForFields：推送的message包含一项sObject字段值的信息，会推送哪些字段值取决于NotifyForFields的赋值，NotifyForFields赋值及作用如下：

• All ：推送所有的字段
• Referenced (default) : 推送select以及where部分字段的并集集合，如果不指定此字段值，默认值为Referenced
• Select ：只推送select部分的字段；
• Where；只推送where部分的字段。

 NotifyForOperationCreate：指定此字段值为true情况下，当操作的sObject进行insert操作，并且满足query的查询条件，则server端会推送一条通知给所有的订阅者们，此字段当且仅当ApiVersion大于29情况下才允许使用；

NotifyForOperationDelete：指定此字段值为true情况下，当操作的sObject进行delete操作，并且满足query的查询条件，则server端会推送一条通知给所有的订阅者们，此字段当且仅当ApiVersion大于29情况下才允许使用；

NotifyForOperationUndelete：指定此字段值为true情况下，当操作的sObject进行undelete操作，并且满足query的查询条件，则server端会推送一条通知给所有的订阅者们，此字段当且仅当ApiVersion大于29情况下才允许使用；

NotifyForOperationUpdate：指定此字段值为true情况下，当操作的sObject进行update操作，并且满足query的查询条件，则server端会推送一条通知给所有的订阅者们，此字段当且仅当ApiVersion大于29情况下才允许使用。

ApiVersion:此字段声明Streaming版本。在36.0及以前，他不包含客户端的状态,也没法跟踪已经过去的事件信息。简单来说，如果声明的版本为36及以前情况下，客户端订阅了渠道后，只能收到订阅后的满足条件的数据事件发生的数据，订阅以前的相关满足事件的数据便没法接收到。如果将此字段设置值为37.0及以上，Streaming API支持存储24小时内满足条件的数据通知信息，即使客户端订阅渠道后，也可以重播24小时内的以前的数据。一条记录有可能出现增删改等很多的事件，每一个广播事件通过replayId作为编号，replayId在org和渠道中是唯一的，即使事件被删除后，此事件对应的replayId也不会被重用，订阅者可以通过赋值ReplayId接收不同的事件消息进行重播，主要有三种ReplayId 赋值

-2：订阅者连接渠道后，接收所有的事件，包括订阅时前24小时的事件消息；

-1：订阅者连接渠道后，接收订阅以后的事件消息；

replayId：订阅者连接渠道后，接受指定replayId以及以后的事件消息，比如replayId为5，订阅者将会受到5以后的事件消息。

(注：每个Query语句中只能搜索一个sObject.)

 

当新建PushTopic想要查看是否创建成功以及是否生效，或者模拟PushTopic的订阅，可以使用workbench查看相关的状态，这里以Account为例，监听Account增删改事件，有以上事件则会发送通知。

 

1）登录workbench,网址https://workbench.developerforce.com/login.php，选择登录的版本为36.0

 

 

2）Jump To选择Streaming Push Topics

 

 

3)Push Topics 选择刚刚创建的TestAccountStreaming，点击Subscribe,则目前已经模拟订阅了此PushTopic

 

 

 

 

 4）修改了一条Account信息，则会在Streaming Push Topic 显示具体的修改信息推送消息了。

 

 

 

 

PushTopic按照上面的字段描述可以很轻松的创建了，下一步应该是让客户端来订阅。如何让订阅呢，这时，我们需要定义一个渠道，保证客户端可以顺利订阅此渠道（Channel），当有满足条件事件的数据进行触发后，由渠道server主动推送给订阅此渠道的客户端。渠道在salesforce端也封装成了一个对象 StreamingChannel，

StreamingChannel表结构如下：

• Description ： Streaming Channel的描述
• ID ： Streaming Channel的ID
• IsDeleted ： 指定Streaming Channel是否被移动到回收站
• IsDynamic ： 如果为true则在订阅时动态创建channel
• LastReferencedDate ： 存储当前用户最近一次查看的这条记录的时间戳
• Name：绑定的PushTopic，以/u/开始，命名为/u/pushtopic Name
• OwnerId : Streaming Channel的owner

这里主要需要说明的一个字段就是Name，Name字段要求必须为/u/PushTopicName，并且PushTopicName只能包含标准的字符以及‘_’或者‘/’，其他的字符不支持，比如上面创建的PushTopic的名称为TestAccountPushTopic，则创建StreamingChannel时，Name必须为/u/TestAccountPushTopic,并且区分大小写。

 2.客户端订阅渠道

客户端订阅渠道可以简单的分成三部分：

• 发送一个握手请求
• 向渠道发送一个订阅的请求
• 使用长响应（轮询）进行连接。

如果订阅是在页面或者组件中使用，可以使用Cometd进行操作，Cometd提供了基本方法用来实现客户端订阅操作，详见Cometd API。当渠道成功后，PushTopic定义的query的数据执行的操作事件（create/update/delete/undelete）满足了PushTopic定义的操作条件，渠道将会向所有订阅的客户端发送通知。上面说道，通知是一个JSON的字符串，大概格式如下所示：

{
  "clientId": "lxdl9o32njygi1gj47kgfaga4k", 
  "data": {
    "event": {
      "createdDate": "2016-09-16T19:45:27.454Z", 
      "replayId": 1, 
      "type": "created"
    }, 
    "sobject": {
      "Phone": "[phone]", 
      "Id": "001D000000KneakIAB", 
      "Name": "Blackbeard"
    }
  }, 
  "channel": "/topic/AccountUpdates"
}

 

订阅者订阅接收消息需要有以下的权限：

where条件部分的字段的FLS

query对象的读的权限

PushTopic对象的读的权限

基于Sharing Rule的新建和编辑记录的权限

Salesforce存储了基于PushTopic时间24小时，并且允许你去重新检索已经存储的以及新的事件。订阅者可以通过重播的选项来选择哪些事件接收。

当你订阅一个渠道后，你并不想接收所有的数据，比如对于客户信息，不同的人员更关注自己的客户的变化信息，这是你可以在订阅的URL后添加filter对事件通知进行过滤，推送你需要的通知信息。

eg：/topic/ChannelName?<expression>

其中ChannelName为上面定义的渠道的名称，<expression>包含一个或者多个条件，多个条件使用\'&\'分割，比如

/topic/TestAccountPushTopic?country=China&isActive=true
如果使用ID作为过滤条件，需要使用18位的ID格式，15位的不支持。

三.代码举例

官方提供了两个demo，一个是基于数据进行DML操作发送给订阅者消息，一个是主动广播发送给订阅者消息。

https://github.com/developerforce/SalesforceDurableStreamingDemo/archive/master.zip

可以下载下来看一下代码，代码里面详尽的讲解了客户端握手，订阅以及接受信息的操作。

总结：本篇主要简单讲解Streaming API的使用方式以及什么情况下会使用到此API，如果使用到查看上面的文档即可。