Is it possible to stream responses from the HTTP processor?

[Zambito1]

The http_client has a stream field that allows you to break the response into a series of messages. I am trying to use Benthos to link sources to Ollama, which has a streaming API. I tried using the unarchive processor, but that will still block until the reply is complete, which is not ideal. Here is what I currently have:

pipeline:
  processors:
    - mapping: |
        root.model = ${LANGUAGE_MODEL}
        root.messages = [
          {
            "role": "user",
            "content": content().string()
          }
        ]
        root.stream = true
    - http:
        url: "http://${OLLAMA_HOST}/api/chat"
        timeout: 15s
    - unarchive:
        format: json_documents

The timeout must be set higher than the total response time from the API as well, which (though I have not tested) I assume is not required with the http_client timeout field. Is it possible to do what I want with the http processor?

[mihaitodor]

Hey @Zambito1 👋 The unarchive processor doesn't get executed before the http processor finishes fetching all the data. I think what you're looking for is to use the http_client input where you can set the request body using the payload field. Then, if you enable streaming, you can set stream.scanner to json_documents: {}. Unfortunately, it's a bit limited in the sense that you can't inject payloads into it which were produced by some other input or processor. If you need something like that, then I think the best way is to consider writing a custom input.