Where communities thrive


  • Join over 1.5M+ people
  • Join over 100K+ communities
  • Free without limits
  • Create your own community
People
Repo info
Activity
  • 08:40
    Andarist opened #6777
  • Jan 19 22:22
    EarthyOrange review_requested #6769
  • Jan 19 16:08
    HyperLife1119 commented #5116
  • Jan 18 05:27
    nodejsdeveloperskh commented #6530
  • Jan 18 05:25
    nodejsdeveloperskh commented #6530
  • Jan 17 17:43
    jennings commented #6501
  • Jan 17 17:43
    jennings commented #6501
  • Jan 17 17:43
    jennings synchronize #6501
  • Jan 17 15:23
    Aidar-Frontend-Developer commented #6775
  • Jan 17 09:20
    jakovljevic-mladen commented #6627
  • Jan 17 09:16
    Aidar-Frontend-Developer opened #6775
  • Jan 17 00:58
    benlesh commented #6769
  • Jan 16 06:52
    EarthyOrange commented #6769
  • Jan 16 06:52
    EarthyOrange commented #6774
  • Jan 16 06:52
    EarthyOrange commented #6774
  • Jan 15 19:53
    kwonoj commented #6774
  • Jan 15 19:52
    dependabot[bot] edited #6773
  • Jan 15 19:51
    dependabot[bot] synchronize #6773
  • Jan 15 19:51
    dependabot[bot] edited #6773
  • Jan 15 19:51

    dependabot[bot] on npm_and_yarn

    chore(deps-dev): bump shelljs f… (compare)

PikaChokeMe
@PikaChokeMe
I feel like I'm confused about where to actually use that operator
map((listOfCommits: any) => listOfCommits.values),

At this point I have an array,
using

mergeMap((listOfCommits: any) => {
          return listOfCommits.map((commit: any) => {

here instead of map just tells me that mergeMap is deprecated and that I should use an inner map instead... I'm not sure what that means or if this is where I was supposed to mergeMap to begin with

Dorus
@Dorus
when you call map<T, T2>(e => httpcall(e)) you're going from Observable<T> -> Observable<Observable<T2>> right?
mergeMap will flatten this (as will concatMap, exhaustMap, switchMap, but in different concurrent patterns). Thus your Observable<Observable<T2>> back to <Observable<T2>.
Just like flatten on an array T[][] will give T[].
@TheeRFG are you using one of the deprecated overloads of mergeMap?
PikaChokeMe
@PikaChokeMe
I think I actually had a misplaced parenthesis which was giving me that issue
Dorus
@Dorus
They deprecated mergeMap(selector, resultSelctor) in favor of mergeMap(selector) https://rxjs-dev.firebaseapp.com/guide/v6/migration#howto-result-selector-migration
@TheeRFG Also using mergeMap on an array will flatten the array. That on itself could be useful, but not the primary purpose of mergeMap. the selector should return an observable that is then merged together with other observables emitted from other runs of the selector.
PikaChokeMe
@PikaChokeMe

so currently with

getLastCommits(): Observable<PRData[]> {
    return this.bitbucketService
      .getLatestCommits()
      .pipe(
        map((listOfCommits: any) => listOfCommits.values),
        tap((whatever: any) => console.log(whatever)),
        concatMap((listOfCommits: any) => {
          return listOfCommits.map((commit: any) => {
            const hash = commit.hash;
            return this.bitbucketService.getPullRequestIDfor(hash)
              .pipe(
                map(id => ({
                          commitHash: hash,
                          pullRequestID: id
                        } as PRData))
              );
          });
        }),
        tap((whatever: any) => console.log(whatever)),
      );
  }

when I call

  ngOnInit(): void {
    this.getLastCommits()
      .subscribe((commitData: PRData[]) => {
        this.commitData = commitData;
        console.log(commitData);
      });
  }

commitData is an observable... subscribing still nets me a single observable?
like commitData is still a single observable

Dorus
@Dorus
try from(listOfCommits) and use mergeMap on the result.
Right now concatmap works on the result of listOfCommits.map, what i think will be an array of observable
PikaChokeMe
@PikaChokeMe
in my previous map(listOfCommits: any) example?
Dorus
@Dorus
In this case i think you want to preserve order? Might make more sense to use forkJoin instead.
getLastCommits(): Observable<PRData[]> {
    return this.bitbucketService
      .getLatestCommits()
      .pipe(
        map((listOfCommits: any) => listOfCommits.values),
        tap((whatever: any) => console.log(whatever)),
        mergeMap((listOfCommits: any) => {
          return forkJoin(...(listOfCommits.map(commit =>
            getPRId(commit.hash)
          )))
        }),
        tap((whatever: any) => console.log(whatever)),
      );
  }

getPRId(hash) {
  return this.bitbucketService.getPullRequestIDfor(hash).pipe(
    map(id => ({
        commitHash: hash,
        pullRequestID: id
      } as PRData))
}
but i'm not sure i understand your types here
PikaChokeMe
@PikaChokeMe
I'll give that a try and if it doesn't work I'll try to better explain my types
I've to submit the project and this seems like a hurdle
Derek
@derekkite
@mkamranhamid how are you calling loadmore? you are passing an observable to a function. check your function parameters
typogod
@typogod
I've made a library that has a very similar API as RxJS but works on pulling instead of pushing. Anyone wanna help me think of a cute name for it?
Tyler Eastman
@teastman
I'm new to rxjs and creating an angular service to cache the current users profile, profile updates can be sent via websocket or periodic refresh. Observables seemed to be a really clean way to allow components to bind the user's profile and send along updates whenever they come in. Searching around I had a hard time finding many examples of this specific pattern, are there any pitfalls I'm missing here?
export class ProfileService {

  private publisher: Subject<Account>;
  public profile: Observable<Account>;

  constructor(private httpClient: HttpClient) {
    this.publisher = new Subject<Account>();
    this.profile = this.publisher.pipe(
      shareReplay(1)
    );
    this.refreshProfile();
  }

  refreshProfile(): void {
    this.httpClient.get<Account>('/profile')
      .pipe(
        map<any, Account>(Account.deserialize)
      ).subscribe(account => this.publisher.next(account), err => this.publisher.error(err));
  }
}
Dorus
@Dorus
@teastman yes, i would recommand the function to return the observable and not subscribe. Right now any error will kill the subject.
if you want to add a store, add it seperate, and just next to a behaviorSubject either from the refresh function with a tap, or have an open websocket that does so.
Calling methods can either subscribe to the behaviorSubject to retrieve the current value, or subscribe to the refresh function to get the latest one, auto updating any other listener.
Tyler Eastman
@teastman
Thank you very much @Dorus that make a lot of sense, appreciate you taking the time.
Naiwei Zheng
@naiweizheng
Is there an easy debugging technique to locate the initiator of a completion/unsubscribe on an observable chain? Going through the call stack is very painful.
Derek
@derekkite

@naiweizheng you can use something like this

tap(
   v => console.log('value', v),
   err => console.log('error', err),
   () => console.log('complete')
)

insert it in the chain and it you can figure out where it is erroring out or completing

lake-toya
@lake-toya

Hey guys,
I'm a little lost right now
I want to build a fairly simple queue.
The queue should only execute one job at a time.
When it is empty it should run a scheduling function (on a x second interval) which checks, if there are new jobs

I was thinking about doing something like this, but it feels a little wrong. Can anyone help me out?

const { of, interval, empty } = require('rxjs')
const { tap, concatMap, skipWhile } = require('rxjs/operators')

const state = {
  isIdle: true,
  pendingJobs: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] // mocked jobs
}

const schedule = () => {
  console.log('scheduling')
  return !!state.isIdle && state.pendingJobs.length
    ? of(state.pendingJobs[0])
    : empty()
}

const executeJob = async (i) => {
  await new Promise(resolve => setTimeout(resolve, 1500))
  return i
}

interval(500).pipe(
  skipWhile(_ => !state.isIdle),
  concatMap(schedule),
  tap(() => { state.isIdle = false }),
  concatMap(executeJob),
  tap((i) => {
    state.pendingJobs = state.pendingJobs.slice(1)
    state.isIdle = true
    console.log('done:', i)
  })
).subscribe()
lake-toya
@lake-toya
Or should I just combine the scheduling and executing into one function and use exhaustMap?
Jorrit
@jorrit-wehelp
sounds to me as pendingJobs.pipe(mergeMap(job => execute(job), MAX_CONCURRENCY)), where pendingJobs does the polling inside and just emits job (or jobs array)
or just concatMap if it's always 1
lake-toya
@lake-toya
Wouldnt pendingJobs keep emitting values even while the execution is still running?
the jobs will be pulled one at a time from a database so I only want to get new jobs when the queue is idling
interval(500).pipe(
  exhaustMap(_ => defer(() => scheduleAndExecute())), // return empty() when no new jobs are found and of(...) when a job was finished
  mergeAll(),
  tap(i => { console.log(i) })
).subscribe()
maybe something like this?
lake-toya
@lake-toya
maybe queue is the wrong term here :D
Jorrit
@jorrit-wehelp
yeah and introducing your own schedule concept is confusing, since rxjs already has its own schedulars :p
But if you get them one at a time (so the 'queue' is in the database I guess), can't you use a repeat() to resubscribe to the db get action after the job has been processed
lake-toya
@lake-toya
oh right, that would be an easy way to do it ... haha thanks
jcapil-ictdu
@jcapil-ictdu
Hello all, any resources for reactjs for beginner. thank you.
Dorus
@Dorus
@jcapil-ictdu wrong channel
bitQ2019
@bitQ2019
hello
PikaChokeMe
@PikaChokeMe
I have an array of Observable<httpData>...
Basicaly Observable<httpData>[]
how do I turn this into an Observable<httpData[]> ?
PikaChokeMe
@PikaChokeMe

I tried to say

    return of(this.jenkinsService.getArrayOfHttpData())
      .pipe(
        mergeMap((observableArray: Observable<httpData>[]) => {
          return forkJoin(...(observableArray.map((observableItem: Observable<httpData>) =>
            observableItem
          )));
        })
      );
  }

But I have like no idea what I'm doing :l

matrixbot
@matrixbot
elpdpt Ryan (Gitter): Not sure I understand your edit becuase its the same, but your solution is right for what you said first. You can simplify it also
return of(this.jenkinsService.getArrayOfHttpData())
  .pipe(
    switchMap(dataObservablesList => forkJoin(...dataObservablesList))
  );
elpdpt * Ryan (Gitter): Your solution is right for what you said first. You can simplify it also
return of(this.jenkinsService.getArrayOfHttpData())
  .pipe(
    switchMap(dataObservablesList => forkJoin(...dataObservablesList))
  );
Dorus
@Dorus
@TheeRFG Try from(arr).pipe(mergeAll())
That will make it emit all httpData's seperate. You could then add a toArray() at the end to change it to Observabel<httpData[]>, but i would probably suggest against that, the idea of RxJs is that you process async streams and handle the results as they come in, not delay till all results are ready.
StephenS
@S-Stephen

Hi, Is someone able to help me figure out what operator(s) I should be using?
I have two observables obs1 and obs2, and I want to combine their latest results only when Obs2 changes something like:

Obs1>----A----B---C----D----E-----
Obs2>--F--------------G---------H--
Out_>-----AF---------CG--------EH

What operator(s) should I use? I've considered using combineLatest, storing the last value of Obs2, and if changes emit the two values, but is this the best approach? (it doesn't feel right using a ?side effect? to track the state) and what operator should I used to 'emit' the combined values?

matrixbot
@matrixbot
elpdpt StephenS (Gitter): It seems you can use distinctUntlChanged with custom compareFunction, checking for the first value in the array for distinction.
distinctUntilChanged((a, b) => a[0] !== b[0])
Dorus
@Dorus
@S-Stephen look at withLatestFrom
looks like
Out_ = Obs2.pipe(withLatestFrom(Obs1))
StephenS
@S-Stephen
@matrixbot @Dorus thanks both. it looks as though withLatestFrom fits the bill
Arun Gopalpuri
@arun0009
How do I forkJoin an observable of array? Do i have to convert it into an array of observable?