javascript - You provided an invalid object where a stream was expected. You can provide an Observable, Promise, Array, or Itera

admin管理员组

文章数量:1287919

Say I have the following function references:

const externalRequests = (params) => Rx.Observable.zip(
  externalApi1(params),
  externalApi2(params)
)

and

const internalDB = (params) => Rx.Observable.fromPromise(getStuffFromDB(params)

externalRequests returns something of the shape:

{ _isScalar: false,
  source: 
   { _isScalar: false,
     source: { _isScalar: false, array: [Object], scheduler: undefined },
     operator: { project: [Object] } },
  operator: undefined }

and can be .subscribe'd to.

internalDB, when .then'd directly returns the right data from the database, and when wrapped in Rx.Observable.fromPromise as above, returns something of the shape:

{ _p: 
   { handle: 115,
     type: 'promise',
     className: 'Object',
     constructorFunction: { ref: 122 },
     protoObject: { ref: 123 },
     prototypeObject: { ref: 1 },
     status: 'pending',
     promiseValue: { ref: 1 },
     properties: [],
     internalProperties: [ [Object], [Object] ],
     text: '#<Promise>' },
  _s: {} }

, which, when .subscribe'd directly returns something like so:

{ isStopped: false,
  observer: 
   { isStopped: false,
     _onNext: [Function: val],
     _onError: [Function: val],
     _onCompleted: [Function: val] },
  m: 
   { isDisposed: false,
     current: { isDisposed: false, current: null } } }

Say I now want to flatten all of this into a single stream: so I write something like:

Rx.Observable.zip(externalRequests, internalDB).mergeAll().subsribe(() => console.log('potato'))

And receive an exception of the form: You provided an invalid object where a stream was expected. You can provide an Observable, Promise, Array, or Iterable.

To summarize: What is the correct way for me to cast my promise as an Observable, and how do I merge/flatten multiple observables as above?

Say I have the following function references:

const externalRequests = (params) => Rx.Observable.zip(
  externalApi1(params),
  externalApi2(params)
)

and

const internalDB = (params) => Rx.Observable.fromPromise(getStuffFromDB(params)

externalRequests returns something of the shape:

{ _isScalar: false,
  source: 
   { _isScalar: false,
     source: { _isScalar: false, array: [Object], scheduler: undefined },
     operator: { project: [Object] } },
  operator: undefined }

and can be .subscribe'd to.

internalDB, when .then'd directly returns the right data from the database, and when wrapped in Rx.Observable.fromPromise as above, returns something of the shape:

{ _p: 
   { handle: 115,
     type: 'promise',
     className: 'Object',
     constructorFunction: { ref: 122 },
     protoObject: { ref: 123 },
     prototypeObject: { ref: 1 },
     status: 'pending',
     promiseValue: { ref: 1 },
     properties: [],
     internalProperties: [ [Object], [Object] ],
     text: '#<Promise>' },
  _s: {} }

, which, when .subscribe'd directly returns something like so:

{ isStopped: false,
  observer: 
   { isStopped: false,
     _onNext: [Function: val],
     _onError: [Function: val],
     _onCompleted: [Function: val] },
  m: 
   { isDisposed: false,
     current: { isDisposed: false, current: null } } }

Say I now want to flatten all of this into a single stream: so I write something like:

Rx.Observable.zip(externalRequests, internalDB).mergeAll().subsribe(() => console.log('potato'))

And receive an exception of the form: You provided an invalid object where a stream was expected. You can provide an Observable, Promise, Array, or Iterable.

To summarize: What is the correct way for me to cast my promise as an Observable, and how do I merge/flatten multiple observables as above?

• javascript
• promise
• rxjs
• observable

Share Improve this question Follow asked Jun 19, 2017 at 14:28 Abraham PAbraham P 15.5k1313 gold badges6363 silver badges130130 bronze badges 1

• 1 My guess is that the problem is in internalDB which is a method returning an Observable. It's not an Observable itself so zip() can't subscribe to a method. Anyway if you don't provide stackoverflow./help/mcve it's hard to tell... – martin Commented Jun 19, 2017 at 14:34

Add a ment  | 

3 Answers 3

Sorted by: Reset to default 3

As @martin mentioned you give functions to the zip operator, not observables.

You need to call those functions with the needed params for your code to work.

Also, I don't think you need the mergeAll call, as the result of internalDB and externalRequests calls are only observables, not higher order observables (observables containing observables)

Rx.Observable
  //    ▼ you need to call the functions in order to get their observables
  .zip(externalRequests(params), internalDB(params))
  .subscribe(() => console.log('potato'))

my mistake to import file like this...

import {loginEpic} from "./Epic";

and i just correct this like...

import loginEpic from "./Epic";

usually it's because you imported zip from 'rxjs/operators'

you should import zip from 'rxjs/observable'

本文标签：