带有 Indy 的 HTTP 连续分组流答案

【问题标题】：HTTP continuous packeted stream with Indy带有 Indy 的 HTTP 连续分组流
【发布时间】：2013-03-14 23:11:18
【问题描述】：

我有一个 JSON-RPC 服务，它针对其中一个请求返回一个连续的 JSON 对象流。

即：

{id:'1'}
{id:'2'}
//30 minutes of no data
{id:'3'}
//...

当然，没有 Content-Length，因为流是无穷无尽的。

我正在使用自定义 TStream 后代来接收和解析数据。但在内部 TIdHttp 缓冲数据并且在收到 RecvBufferSize 字节之前不会将其传递给我。

这会导致：

{id:'1'} //received
{id:'2'} //buffered by Indy but not received
//30 minutes of no data
{id:'3'} //this is where Indy commits {id:'2'} to me

显然这不行，因为 30 分钟前重要的消息应该在 30 分钟前送达。

我希望 Indy 做套接字所做的事情：如果有可用数据，则读取到 RecvBufferSize 或更少并立即返回。

我发现 this discussion 从 2005 年开始，一些可怜的人试图向 Indy 开发人员解释这个问题，但他们不理解他。（阅读它；这是一个悲伤的景象）

无论如何，他通过编写自定义 IOHandler 后代来解决这个问题，但那是在 2005 年，也许今天有一些现成的解决方案？

【问题讨论】：

标签： delphi indy indy10 idhttp

【解决方案1】：

您不需要编写 IOHandler 后代，TIdTCPClient 类已经可以。它公开了一个TIdIOHandler 对象，该对象具有从套接字读取的方法。这些 ReadXXX 方法会一直阻塞，直到请求的数据被读取或发生超时。只要连接存在，ReadXXX 就可以循环执行，每当它接收到新的 JSON 对象时，将其传递给应用程序逻辑。

您的示例看起来所有 JSON 对象只有一行。然而，JSON 对象可以是多行的，在这种情况下，客户端代码需要知道它们是如何分开的。

更新：在类似 Stackoverflow 问题（针对 .Net）中，针对“流式”HTTP JSON Web 服务，最受好评的解决方案使用较低级别的 TCP 客户端而不是 HTTP 客户端：Reading data from an open HTTP stream

【讨论】：

【解决方案2】：

在我看来，这就像一个 WebSocket 任务，因为您的连接不再是面向普通的 HTTP 问题/答案，而是内容流。

查看WebSocket server implementations for Delphi 获取一些代码。

有at least one based on Indy，来自AsmProfiler的作者。

AFAIK 在 websockets 中有两种流：二进制和文本。从 websocket 的角度来看，我怀疑您的 JSON 流是一些文本内容。

另一种选择是使用long-pooling 或一些更旧的协议，它们对rooter 更友好——当连接切换到websockets 模式时，它不再是标准的HTTP，所以一些“明智的”数据包检查工具（在公司网络）可能会将其识别为安全攻击（例如 DoS），因此可能会停止连接。

【讨论】：

如果我做对了，这两种解决方案都需要重写服务吗？因为我无权访问它。
@himself 如果您的请求是打开连接并且不使用 Content-Length 标头，这不再是 HTTP，因此我想您将不得不更改服务端！
嗯，猜猜服务端会说什么？ “HTTP 标准中没有任何地方说 HTTP 中间件可以长时间缓冲数据。因此我们的服务很好，我想你必须修复你的 HTTP 客户端代码”。回到第一格。
@ArnaudBouchez 它仍然是 HTTP - 可以省略 content-length 标头，服务器只是在最后关闭连接（请参阅RFC 2616），默认情况下连接在 HTTP 1.1 中是持久的
@mjn 我的意思是，即使它“可能”符合 HTTP 标准，websockets 布局对于某些 OSI 7 级安全检查员来说也不好闻。如果服务器或路由器没有将流识别为“经典”HTTP 无状态进程，它可能会以某种超时停止连接。这可能是 DoS 的大门 - 请参阅 bugs.webkit.org/show_bug.cgi?id=32246 和 media.blackhat.com/bh-us-12/Briefings/Shekyan/…

【解决方案3】：

虽然使用 TCP 流是一种选择，但最后我还是采用了编写自定义 TIdIOHandlerStack 后代的原始解决方案。

动机是使用 TIdHTTP 我知道什么不起作用，只需要修复它，而切换到较低级别的 TCP 意味着可能会出现新问题。

Here's the code that I'm using，我这里就重点讨论一下。

新的TIdStreamIoHandler 必须继承自TIdIOHandlerStack。

两个函数需要重写：ReadBytes和ReadStream：

function TryReadBytes(var VBuffer: TIdBytes; AByteCount: Integer;
  AAppend: Boolean = True): integer; virtual;
procedure ReadStream(AStream: TStream; AByteCount: TIdStreamSize = -1;
  AReadUntilDisconnect: Boolean = False); override;

两者都是经过修改的 Indy 函数，可以在 IdIOHandler.TIdIOHandler 中找到。在ReadBytes 中，必须将while 子句替换为单个ReadFromSource() 请求，以便TryReadBytes 在一次读取最多AByteCount 个字节后返回。

基于此，ReadStream 必须处理 AByteCount (>0,

注意ReadStream 不需要提前终止，即使在这个流版本中，如果只有部分请求的数据在套接字中可用。它只需要立即将该部分写入流而不是将其缓存在FInputBuffer 中，然后阻塞并等待下一部分数据。

【讨论】：

由于 Indy 是开源的，修改后的源可以（并且，如果对其他人有帮助，应该）公开

【解决方案4】：

在分块编码传输模式下传输的数据包内容之前实际上有一个长度数据。使用这个长度数据，idhttp 的 IOhandler 一个包一个包地读取流。最小有意义的单位是一个数据包，所以不需要从一个数据包中一个一个地读取字符，也不需要改变IOHandler的功能。唯一的问题是 idhttp 不会停止将流数据转到下一步，因为流数据无穷无尽：没有结束数据包。因此解决方案是使用 idhttp onwork 事件触发从流中读取并将流位置设置为零以避免溢出。像这样：

    //add a event handler to idhttp    
    IdHTTP.OnWork :=  IdHTTPWork;


    procedure  TRatesStreamWorker.IdHTTPWork(ASender: TObject; AWorkMode: TWorkMode; AWorkCount: Int64);
    begin
         .....
         ResponseStringStream.Position :=0; 
         s:=ResponseStringStream.ReadString(ResponseStringStream.Size) ;//this is the packet conten
         ResponseStringStream.Clear;
         ... 
    end;

procedure TForm1.ButtonGetStreamPricesClick(Sender: TObject);
var
 begin
    .....    
    source := RatesWorker.RatesURL+'EUR_USD';  
    RatesWorker.IdHTTP.Get(source,RatesWorker.ResponseStringStream);  
 end;

然而，使用 Tstream 的自定义 write() 函数可能是这种需求的更好解决方案。

【讨论】：