Companion stream upload unknown size files

docs/companion.md

@@ -329,6 +329,7 @@ const options = {
 	logClientVersion: true,
 	periodicPingUrls: [],
 	streamingUpload: true,
+	streamingUploadSizeless: false,
 	clientSocketConnectTimeout: 60000,
 	metrics: true,
 };
@@ -640,6 +641,15 @@ enabled, it will lead to _faster uploads_ because companion will start uploading
 at the same time as downloading using `stream.pipe`. If `false`, files will be
 fully downloaded first, then uploaded. Defaults to `true`.
 
+#### `streamingUploadSizeless` `COMPANION_STREAMING_UPLOAD_SIZELESS`
+
+A boolean flag to tell Companion whether to also upload files that have an
+unknown size. Currently this is only supported for Tus uploads. Note that this
+requires an optional extension on the Tus server if using Tus uploads. For form
+multipart uploads it requres a server that can handle
+`transfer-encoding: chunked`. Default is `false`. If set to `true`,
+`streamingUpload` also has to be set to `true`.
+
 #### `maxFileSize` `COMPANION_MAX_FILE_SIZE`
 
 If this value is set, companion will limit the maximum file size to process. If

docs/framework-integrations/react.mdx

@@ -74,7 +74,7 @@ times, this is needed if you are building a custom UI for Uppy in React.
 const [uppy] = useState(() => new Uppy());
 
 const files = useUppyState(uppy, (state) => state.files);
-const totalProgress = useUppyState(uppy, (state) => state.totalProgress);
+const totalProgress = useUppyState(uppy, (state) => state.progress);
 // We can also get specific plugin state.
 // Note that the value on `plugins` depends on the `id` of the plugin.
 const metaFields = useUppyState(

docs/uppy-core.mdx

@@ -794,7 +794,7 @@ const state = {
 	capabilities: {
 		resumableUploads: false,
 	},
-	totalProgress: 0,
+	progress: null,
 	meta: { ...this.opts.meta },
 	info: {
 		isHidden: true,

e2e/start-companion-with-load-balancer.mjs

@@ -1,9 +1,10 @@
 #!/usr/bin/env node
 
-import { spawn } from 'node:child_process'
 import http from 'node:http'
 import httpProxy from 'http-proxy'
 import process from 'node:process'
+import { execaNode } from 'execa';
+
 
 const numInstances = 3
 const lbPort = 3020
@@ -49,41 +50,27 @@ function createLoadBalancer (baseUrls) {
 const isWindows = process.platform === 'win32'
 const isOSX = process.platform === 'darwin'
 
-const startCompanion = ({ name, port }) => {
-  const cp = spawn(process.execPath, [
+const startCompanion = ({ name, port }) => execaNode('packages/@uppy/companion/src/standalone/start-server.js', {
+  nodeOptions: [
     '-r', 'dotenv/config',
     // Watch mode support is limited to Windows and macOS at the time of writing.
     ...(isWindows || isOSX ? ['--watch-path', 'packages/@uppy/companion/src', '--watch'] : []),
-    './packages/@uppy/companion/src/standalone/start-server.js',
-  ], {
-    cwd: new URL('../', import.meta.url),
-    stdio: 'inherit',
-    env: {
-      // Note: these env variables will override anything set in .env
-      ...process.env,
-      COMPANION_PORT: port,
-      COMPANION_SECRET: 'development', // multi instance will not work without secret set
-      COMPANION_PREAUTH_SECRET: 'development', // multi instance will not work without secret set
-      COMPANION_ALLOW_LOCAL_URLS: 'true',
-      COMPANION_ENABLE_URL_ENDPOINT: 'true',
-      COMPANION_LOGGER_PROCESS_NAME: name,
-      COMPANION_CLIENT_ORIGINS: 'true',
-    },
-  })
-  // Adding a `then` property so the return value is awaitable:
-  return Object.defineProperty(cp, 'then', {
-    __proto__: null,
-    writable: true,
-    configurable: true,
-    value: Promise.prototype.then.bind(new Promise((resolve, reject) => {
-      cp.on('exit', (code) => {
-        if (code === 0) resolve(cp)
-        else reject(new Error(`Non-zero exit code: ${code}`))
-      })
-      cp.on('error', reject)
-    })),
-  })
-}
+  ],
+  cwd: new URL('../', import.meta.url),
+  stdio: 'inherit',
+  env: {
+    // Note: these env variables will override anything set in .env
+    ...process.env,
+    COMPANION_PORT: port,
+    COMPANION_SECRET: 'development', // multi instance will not work without secret set
+    COMPANION_PREAUTH_SECRET: 'development', // multi instance will not work without secret set
+    COMPANION_ALLOW_LOCAL_URLS: 'true',
+    COMPANION_ENABLE_URL_ENDPOINT: 'true',
+    COMPANION_LOGGER_PROCESS_NAME: name,
+    COMPANION_CLIENT_ORIGINS: 'true',
+  },
+})
+
 
 const hosts = Array.from({ length: numInstances }, (_, index) => {
   const port = companionStartPort + index

examples/react-example/App.tsx

@@ -77,7 +77,7 @@ export default function App() {
     uppy,
     (state) => Object.keys(state.files).length,
   )
-  const totalProgress = useUppyState(uppy, (state) => state.totalProgress)
+  const totalProgress = useUppyState(uppy, (state) => state.progress)
   // Also possible to get the state of all plugins.
   const plugins = useUppyState(uppy, (state) => state.plugins)
 

examples/react-native-expo/App.js

@@ -39,7 +39,7 @@ export default function App () {
       setState({
         progress: progress.bytesUploaded,
         total: progress.bytesTotal,
-        totalProgress: uppy.state.totalProgress,
+        totalProgress: uppy.state.progress,
         uploadStarted: true,
       })
     })

package.json

@@ -84,6 +84,7 @@
     "eslint-plugin-react": "^7.22.0",
     "eslint-plugin-react-hooks": "^4.2.0",
     "eslint-plugin-unicorn": "^53.0.0",
+    "execa": "^9.5.1",
     "github-contributors-list": "^1.2.4",
     "glob": "^8.0.0",
     "jsdom": "^24.0.0",

packages/@uppy/companion-client/src/RequestClient.ts

@@ -4,7 +4,6 @@ import pRetry, { AbortError } from 'p-retry'
 
 import fetchWithNetworkError from '@uppy/utils/lib/fetchWithNetworkError'
 import ErrorWithCause from '@uppy/utils/lib/ErrorWithCause'
-import emitSocketProgress from '@uppy/utils/lib/emitSocketProgress'
 import getSocketHost from '@uppy/utils/lib/getSocketHost'
 
 import type Uppy from '@uppy/core'
@@ -81,6 +80,26 @@ async function handleJSONResponse<ResJson>(res: Response): Promise<ResJson> {
   throw new HttpError({ statusCode: res.status, message: errMsg })
 }
 
+function emitSocketProgress(
+  uploader: { uppy: Uppy<any, any> },
+  progressData: {
+    progress: string // pre-formatted percentage number as a string
+    bytesTotal: number
+    bytesUploaded: number
+  },
+  file: UppyFile<any, any>,
+): void {
+  const { progress, bytesUploaded, bytesTotal } = progressData
+  if (progress) {
+    uploader.uppy.log(`Upload progress: ${progress}`)
+    uploader.uppy.emit('upload-progress', file, {
+      uploadStarted: file.progress.uploadStarted ?? 0,
+      bytesUploaded,
+      bytesTotal,
+    })
+  }
+}
+
 export default class RequestClient<M extends Meta, B extends Body> {
   static VERSION = packageJson.version
 

packages/@uppy/companion/src/config/companion.js

@@ -20,6 +20,7 @@ const defaultOptions = {
   allowLocalUrls: false,
   periodicPingUrls: [],
   streamingUpload: true,
+  streamingUploadSizeless: false,
   clientSocketConnectTimeout: 60000,
   metrics: true,
 }

packages/@uppy/companion/src/server/Uploader.js

@@ -220,10 +220,14 @@ class Uploader {
     if (this.readStream) this.readStream.destroy(err)
   }
 
-  async _uploadByProtocol(req) {
+  _getUploadProtocol() {
     // todo a default protocol should not be set. We should ensure that the user specifies their protocol.
     // after we drop old versions of uppy client we can remove this
-    const protocol = this.options.protocol || PROTOCOLS.multipart
+    return this.options.protocol || PROTOCOLS.multipart
+  }
+
+  async _uploadByProtocol(req) {
+    const protocol = this._getUploadProtocol()
 
     switch (protocol) {
       case PROTOCOLS.multipart:
@@ -264,8 +268,12 @@ class Uploader {
     this.readStream = fileStream
   }
 
-  _needDownloadFirst() {
-    return !this.options.size || !this.options.companionOptions.streamingUpload
+  _canStream() {
+    return this.options.companionOptions.streamingUpload && (
+      this.options.size
+      // only tus uploads can be streamed without size, TODO: add also others
+      || this.options.companionOptions.streamingUploadSizeless
+    )
   }
 
   /**
@@ -281,7 +289,8 @@ class Uploader {
       this.#uploadState = states.uploading
 
       this.readStream = stream
-      if (this._needDownloadFirst()) {
+
+      if (!this._canStream()) {
         logger.debug('need to download the whole file first', 'controller.get.provider.size', this.shortToken)
         // Some streams need to be downloaded entirely first, because we don't know their size from the provider
         // This is true for zoom and drive (exported files) or some URL downloads.
@@ -429,7 +438,7 @@ class Uploader {
     // If fully downloading before uploading, combine downloaded and uploaded bytes
     // This will make sure that the user sees half of the progress before upload starts (while downloading)
     let combinedBytes = bytesUploaded
-    if (this._needDownloadFirst()) {
+    if (!this._canStream()) {
       combinedBytes = Math.floor((combinedBytes + (this.downloadedBytes || 0)) / 2)
     }
 
@@ -606,7 +615,7 @@ class Uploader {
 
       const response = await runRequest(url, reqOptions)
 
-      if (bytesUploaded !== this.size) {
+      if (this.size != null && bytesUploaded !== this.size) {
         const errMsg = `uploaded only ${bytesUploaded} of ${this.size} with status: ${response.statusCode}`
         logger.error(errMsg, 'upload.multipart.mismatch.error')
         throw new Error(errMsg)

packages/@uppy/companion/src/standalone/helper.js

@@ -182,6 +182,7 @@ const getConfigFromEnv = () => {
     // cookieDomain is kind of a hack to support distributed systems. This should be improved but we never got so far.
     cookieDomain: process.env.COMPANION_COOKIE_DOMAIN,
     streamingUpload: process.env.COMPANION_STREAMING_UPLOAD ? process.env.COMPANION_STREAMING_UPLOAD === 'true' : undefined,
+    streamingUploadSizeless: process.env.COMPANION_STREAMING_UPLOAD_SIZELESS ? process.env.COMPANION_STREAMING_UPLOAD_SIZELESS === 'true' : undefined,
     maxFileSize: process.env.COMPANION_MAX_FILE_SIZE ? parseInt(process.env.COMPANION_MAX_FILE_SIZE, 10) : undefined,
     chunkSize: process.env.COMPANION_CHUNK_SIZE ? parseInt(process.env.COMPANION_CHUNK_SIZE, 10) : undefined,
     clientSocketConnectTimeout: process.env.COMPANION_CLIENT_SOCKET_CONNECT_TIMEOUT