通过 HTTP 将数据从浏览器流式传输到服务器的方法

Question

是否有任何类似 XHR 的浏览器 API 可用于通过 HTTP 将二进制流式传输到服务器？

我想随着时间的推移发出 HTTP PUT 请求并以编程方式创建数据。我不想一次创建所有这些数据，因为内存中可能存在大量数据。一些伪代码来说明我的意思：

var dataGenerator = new DataGenerator(); // Generates 8KB UInt8Array every second
var streamToWriteTo;
http.put('/example', function (requestStream) {
  streamToWriteTo = requestStream;
});

dataGenerator.on('data', function (chunk) {
  if (!streamToWriteTo) {
    return;
  }
  streamToWriteTo.write(chunk);
});

我目前有一个 Web 套接字解决方案，但我更喜欢常规 HTTP，以便与一些现有的服务器端代码更好地互操作。

编辑：我可以使用最前沿的浏览器 API。我正在查看 Fetch API，因为它支持请求正文的 ArrayBuffers、DataViews、Files 等。如果我能以某种方式伪造这些对象之一，以便我可以将 Fetch API 与动态数据一起使用，那将适用于我。我尝试创建一个代理对象，看看是否调用了我可以修补的方法。不幸的是，浏览器（至少在 Chrome 中）似乎正在阅读本机代码，而不是在 JS 领域。但是，如果我错了，请纠正我。

Answer 1

我不知道如何使用纯 HTML5 API 来做到这一点，但一种可能的解决方法是使用 Chrome 应用程序作为后台服务来为网页提供附加功能。如果您已经愿意使用开发浏览器并启用实验性功能，那么这似乎只是进一步的增量步骤。

Chrome 应用程序可以调用chrome.sockets.tcp API，您可以在该 API 上实现您想要的任何协议，包括 HTTP 和 HTTPS。这将为实现流式传输提供灵活性。

普通网页可以使用chrome.runtime API 与应用程序交换消息，只要应用程序declares this usage。这将允许您的网页对您的应用进行异步调用。

我编写了这个简单的应用程序作为概念证明：

manifest.json

{
  "manifest_version" : 2,

  "name" : "Streaming Upload Test",
  "version" : "0.1",

  "app": {
    "background": {
      "scripts": ["background.js"]
    }
  },

  "externally_connectable": {
    "matches": ["*://localhost/*"]
  },

  "sockets": {
    "tcp": {
      "connect": "*:*"
    }
  },

  "permissions": [
  ]
}

background.js

var mapSocketToPort = {};

chrome.sockets.tcp.onReceive.addListener(function(info) {
  var port = mapSocketToPort[info.socketId];
  port.postMessage(new TextDecoder('utf-8').decode(info.data));
});

chrome.sockets.tcp.onReceiveError.addListener(function(info) {
  chrome.sockets.tcp.close(info.socketId);
  var port = mapSocketToPort[info.socketId];
  port.postMessage();
  port.disconnect();
  delete mapSocketToPort[info.socketId];
});

// Promisify socket API for easier operation sequencing.
// TODO: Check for error and reject.
function socketCreate() {
  return new Promise(function(resolve, reject) {
    chrome.sockets.tcp.create({ persistent: true }, resolve);
  });
}

function socketConnect(s, host, port) {
  return new Promise(function(resolve, reject) {
    chrome.sockets.tcp.connect(s, host, port, resolve);
  });
}

function socketSend(s, data) {
  return new Promise(function(resolve, reject) {
    chrome.sockets.tcp.send(s, data, resolve);
  });
}

chrome.runtime.onConnectExternal.addListener(function(port) {
  port.onMessage.addListener(function(msg) {
    if (!port.state) {
      port.state = msg;

      port.chain = socketCreate().then(function(info) {
        port.socket = info.socketId;
        mapSocketToPort[port.socket] = port;
        return socketConnect(port.socket, 'httpbin.org', 80);
      }).then(function() {
        // TODO: Layer TLS if needed.
      }).then(function() {
        // TODO: Build headers from the request.
        // TODO: Use Transfer-Encoding: chunked.
        var headers =
            'PUT /put HTTP/1.0\r\n' +
            'Host: httpbin.org\r\n' +
            'Content-Length: 17\r\n' +
            '\r\n';
        return socketSend(port.socket, new TextEncoder('utf-8').encode(headers).buffer);
      });
    }
    else {
      if (msg) {
        port.chain = port.chain.then(function() {
          // TODO: Use chunked encoding.
          return socketSend(port.socket, new TextEncoder('utf-8').encode(msg).buffer);
        });
      }
    }
  });
});

此应用程序没有用户界面。它侦听连接并向http://httpbin.org/put 发出硬编码的PUT 请求（httpbin 是一个有用的测试站点，但请注意does not support chunked encoding）。 PUT 数据（目前硬编码为 17 个八位字节）从客户端流入（根据需要使用尽可能少或尽可能多的消息）并发送到服务器。来自服务器的响应流回客户端。

这只是一个概念证明。一个真正的应用程序可能应该：

• 连接到任何主机和端口。
• 使用传输编码：分块。
• 表示流数据结束。
• 处理套接字错误。
• 支持 TLS（例如 Forge）

这是一个使用应用即服务执行流式上传（17 个八位字节）的示例网页（请注意，您必须配置自己的应用 ID）：

<pre id="result"></pre>
<script>
 var MY_CHROME_APP_ID = 'omlafihmmjpklmnlcfkghehxcomggohk';

 function streamingUpload(url, options) {
   // Open a connection to the Chrome App. The argument must be the 
   var port = chrome.runtime.connect(MY_CHROME_APP_ID);

   port.onMessage.addListener(function(msg) {
     if (msg)
       document.getElementById("result").textContent += msg;
     else
       port.disconnect();
   });

   // Send arguments (must be JSON-serializable).
   port.postMessage({
     url: url,
     options: options
   });

   // Return a function to call with body data.
   return function(data) {
     port.postMessage(data);
   };
 }

 // Start an upload.
 var f = streamingUpload('https://httpbin.org/put', { method: 'PUT' });

 // Stream data a character at a time.
 'how now brown cow'.split('').forEach(f);
</script>

当我在安装了应用程序的 Chrome 浏览器中加载此网页时，httpbin 返回：

HTTP/1.1 200 OK
Server: nginx
Date: Sun, 19 Jun 2016 16:54:23 GMT
Content-Type: application/json
Content-Length: 240
Connection: close
Access-Control-Allow-Origin: *
Access-Control-Allow-Credentials: true

{
  "args": {}, 
  "data": "how now brown cow", 
  "files": {}, 
  "form": {}, 
  "headers": {
    "Content-Length": "17", 
    "Host": "httpbin.org"
  }, 
  "json": null, 
  "origin": "[redacted]", 
  "url": "http://httpbin.org/put"
}

Answer 2

我目前正在寻找完全相同的东西（通过 Ajax 上传）。 我目前的发现，看起来好像我们正在搜索浏览器功能设计的最前沿；-)

XMLHttpRequest definition 在第 4 步中告诉 bodyinit 的内容提取是（或可以是） readablestream.

我仍在（作为非网络开发人员）搜索有关如何创建这样的东西并将数据馈送到“可读流”的“另一端”的信息（即应该是“可写流”，但我还没有没找到）。

如果您找到了实施这些设计计划的方法，也许您在搜索方面做得更好，并且可以在此处发布。

^5
斯文

Answer 3

一种利用ReadableStream 流式传输任意数据的方法； RTCDataChannel 以Uint8Array 的形式发送和接收任意数据； TextEncoder 创建8000 存储在Uint8Array 中的随机数据字节，TextDecoder 将RTCDataChannel 返回的Uint8Array 解码为字符串进行演示，注意也可以使用FileReader .readAsArrayBuffer 和@987654328 @ 这里。

标记和脚本代码是根据MDN - WebRTC: Simple RTCDataChannel sample 的示例修改的，包括adapter.js，其中包含RTCPeerConnection 助手； Creating your own readable stream.

另请注意，当传输的总字节数达到 8000 * 8 时，示例流被取消：64000

(function init() {
  var interval, reader, stream, curr, len = 0,
    totalBytes = 8000 * 8,
    data = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789",
    randomData = function randomData() {
      var encoder = new TextEncoder();
      var currentStream = "";
      for (var i = 0; i < 8000; i++) {
        currentStream += data[Math.floor(Math.random() * data.length)]
      }
      return encoder.encode(currentStream)
    },
    // optionally reconnect to stream if cancelled
    reconnect = function reconnect() {
      connectButton.disabled = false;
      startup()
    };

  // Define "global" variables

  var connectButton = null;
  var disconnectButton = null;
  var messageInputBox = null;
  var receiveBox = null;

  var localConnection = null; // RTCPeerConnection for our "local" connection
  // adjust this to remote address; or use `ServiceWorker` `onfetch`; other
  var remoteConnection = null; // RTCPeerConnection for the "remote"

  var sendChannel = null; // RTCDataChannel for the local (sender)
  var receiveChannel = null; // RTCDataChannel for the remote (receiver)

  // Functions

  // Set things up, connect event listeners, etc.

  function startup() {
    connectButton = document.getElementById("connectButton");
    disconnectButton = document.getElementById("disconnectButton");
    messageInputBox = document.getElementById("message");
    receiveBox = document.getElementById("receivebox");

    // Set event listeners for user interface widgets

    connectButton.addEventListener("click", connectPeers, false);
    disconnectButton.addEventListener("click", disconnectPeers, false);
  }

  // Connect the two peers. Normally you look for and connect to a remote
  // machine here, but we"re just connecting two local objects, so we can
  // bypass that step.

  function connectPeers() {
    // Create the local connection and its event listeners
    if (len < totalBytes) {
      localConnection = new RTCPeerConnection();

      // Create the data channel and establish its event listeners
      sendChannel = localConnection.createDataChannel("sendChannel");
      sendChannel.onopen = handleSendChannelStatusChange;
      sendChannel.onclose = handleSendChannelStatusChange;

      // Create the remote connection and its event listeners

      remoteConnection = new RTCPeerConnection();
      remoteConnection.ondatachannel = receiveChannelCallback;

      // Set up the ICE candidates for the two peers

      localConnection.onicecandidate = e => 
        !e.candidate || remoteConnection.addIceCandidate(e.candidate)
      .catch(handleAddCandidateError);

      remoteConnection.onicecandidate = e => 
        !e.candidate || localConnection.addIceCandidate(e.candidate)
      .catch(handleAddCandidateError);

      // Now create an offer to connect; this starts the process

      localConnection.createOffer()
      .then(offer => localConnection.setLocalDescription(offer))
      .then(() => remoteConnection
                 .setRemoteDescription(localConnection.localDescription)
       )
      .then(() => remoteConnection.createAnswer())
      .then(answer => remoteConnection
                      .setLocalDescription(answer)
       )
      .then(() => localConnection
                 .setRemoteDescription(remoteConnection.localDescription)
      )
      // start streaming connection
      .then(sendMessage)
      .catch(handleCreateDescriptionError);
    } else {

      alert("total bytes streamed:" + len)
    }

  }

  // Handle errors attempting to create a description;
  // this can happen both when creating an offer and when
  // creating an answer. In this simple example, we handle
  // both the same way.

  function handleCreateDescriptionError(error) {
    console.log("Unable to create an offer: " + error.toString());
  }

  // Handle successful addition of the ICE candidate
  // on the "local" end of the connection.

  function handleLocalAddCandidateSuccess() {
    connectButton.disabled = true;
  }

  // Handle successful addition of the ICE candidate
  // on the "remote" end of the connection.

  function handleRemoteAddCandidateSuccess() {
    disconnectButton.disabled = false;
  }

  // Handle an error that occurs during addition of ICE candidate.

  function handleAddCandidateError() {
    console.log("Oh noes! addICECandidate failed!");
  }

  // Handles clicks on the "Send" button by transmitting
  // a message to the remote peer.

  function sendMessage() {

    stream = new ReadableStream({
      start(controller) {
          interval = setInterval(() => {
            if (sendChannel) {
              curr = randomData();
              len += curr.byteLength;
              // queue current stream
              controller.enqueue([curr, len, sendChannel.send(curr)]);

              if (len >= totalBytes) {
                controller.close();
                clearInterval(interval);
              }
            }
          }, 1000);
        },
        pull(controller) {
          // do stuff during stream
          // call `releaseLock()` if `diconnect` button clicked
          if (!sendChannel) reader.releaseLock();
        },
        cancel(reason) {
          clearInterval(interval);
          console.log(reason);
        }
    });

    reader = stream.getReader({
      mode: "byob"
    });

    reader.read().then(function process(result) {
        if (result.done && len >= totalBytes) {
          console.log("Stream done!");
          connectButton.disabled = false;
          if (len < totalBytes) reconnect();
          return;
        }

        if (!result.done && result.value) {
          var [currentStream, totalStreamLength] = [...result.value];
        }

        if (result.done && len < totalBytes) {
          throw new Error("stream cancelled")
        }

        console.log("currentStream:", currentStream
                   , "totalStremalength:", totalStreamLength
                   , "result:", result);
        return reader.read().then(process);
      })
      .catch(function(err) {
        console.log("catch stream cancellation:", err);
        if (len < totalBytes) reconnect()
      });

    reader.closed.then(function() {
      console.log("stream closed")
    })

  }

  // Handle status changes on the local end of the data
  // channel; this is the end doing the sending of data
  // in this example.

  function handleSendChannelStatusChange(event) {
    if (sendChannel) {
      var state = sendChannel.readyState;

      if (state === "open") {
        disconnectButton.disabled = false;
        connectButton.disabled = true;
      } else {
        connectButton.disabled = false;
        disconnectButton.disabled = true;
      }
    }
  }

  // Called when the connection opens and the data
  // channel is ready to be connected to the remote.

  function receiveChannelCallback(event) {
    receiveChannel = event.channel;
    receiveChannel.onmessage = handleReceiveMessage;
    receiveChannel.onopen = handleReceiveChannelStatusChange;
    receiveChannel.onclose = handleReceiveChannelStatusChange;
  }

  // Handle onmessage events for the receiving channel.
  // These are the data messages sent by the sending channel.

  function handleReceiveMessage(event) {
    var decoder = new TextDecoder();
    var data = decoder.decode(event.data);
    var el = document.createElement("p");
    var txtNode = document.createTextNode(data);

    el.appendChild(txtNode);
    receiveBox.appendChild(el);
  }

  // Handle status changes on the receiver"s channel.

  function handleReceiveChannelStatusChange(event) {
    if (receiveChannel) {
      console.log("Receive channel's status has changed to " +
        receiveChannel.readyState);
    }

    // Here you would do stuff that needs to be done
    // when the channel"s status changes.
  }

  // Close the connection, including data channels if they"re open.
  // Also update the UI to reflect the disconnected status.

  function disconnectPeers() {

    // Close the RTCDataChannels if they"re open.

    sendChannel.close();
    receiveChannel.close();

    // Close the RTCPeerConnections

    localConnection.close();
    remoteConnection.close();

    sendChannel = null;
    receiveChannel = null;
    localConnection = null;
    remoteConnection = null;

    // Update user interface elements


    disconnectButton.disabled = true;
    // cancel stream on `click` of `disconnect` button, 
    // pass `reason` for cancellation as parameter
    reader.cancel("stream cancelled");
  }

  // Set up an event listener which will run the startup
  // function once the page is done loading.

  window.addEventListener("load", startup, false);
})();

plnkr http://plnkr.co/edit/cln6uxgMZwE2EQCfNXFO?p=preview

Answer 4

我认为简短的回答是否定的。截至撰写此回复时（2021 年 11 月），这在任何主要浏览器中均不可用。

长答案是：

我认为您正在使用 Fetch API 寻找正确的位置。 ReadableStream 目前是 Request 构造函数的 body 属性的有效类型：
https://developer.mozilla.org/en-US/docs/Web/API/Request/Request#parameters

但是，如果您查看浏览器支持矩阵，可悲的是：
https://developer.mozilla.org/en-US/docs/Web/API/Request/Request#browser_compatibility
您可以看到“在请求正文中发送 ReadableStream”对于所有主要浏览器仍然为否。虽然它目前在某些浏览器（包括 Chrome）中以实验模式提供。

这里有一个很好的关于如何在实验模式下进行操作的教程：
https://web.dev/fetch-upload-streaming/

查看帖子的日期和在此功能上完成的工作，我认为这项技术很明显处于停滞状态，我们可能不会很快看到它。因此，遗憾的是，WebSockets 可能仍然是我们为数不多的好选择之一（用于无限制的流传输）：
https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API

Answer 5

您可以使用Promise、setTimeout、递归。另见PUT vs POST in REST

var count = 0, total = 0, timer = null, d = 500, stop = false, p = void 0
, request = function request () {
              return new XMLHttpRequest()
            };
function sendData() {
  p = Promise.resolve(generateSomeBinaryData()).then(function(data) { 
    var currentRequest = request();
    currentRequest.open("POST", "http://example.com");
    currentRequest.onload = function () {
      ++count; // increment `count`
      total += data.byteLength; // increment total bytes posted to server
    }

    currentRequest.onloadend = function () {
      if (stop) { // stop recursion
        throw new Error("aborted") // `throw` error to `.catch()`
      } else {
        timer = setTimeout(sendData, d); // recursively call `sendData`
      }
    }
    currentRequest.send(data); // `data`: `Uint8Array`; `TypedArray`
    return currentRequest; // return `currentRequest` 
  });
  return p // return `Promise` : `p`
}

var curr = sendData();

curr.then(function(current) {
  console.log(current) // current post request
})
.catch(function(err) {
  console.log(e) // handle aborted `request`; errors
});

Answer 6

Server-Sent Events 和 WebSockets 是首选方法，但在您的情况下，您希望创建代表性状态传输、REST、API 并使用长轮询。见How do I implement basic “Long Polling”?

在客户端和服务器端都处理长轮询过程。服务器脚本和 http 服务器必须配置为支持长轮询。

除了长轮询，短轮询 (XHR/AJAX) 需要浏览器轮询服务器。