I am exploring the world of RxJS and seeking guidance from experienced individuals.
My goal is to establish a synchronized flow of responses, along with their corresponding requests, from a stream of payload data.
The desired approach involves sending each request sequentially, with every subsequent request awaiting the response from its predecessor.
Although my initial attempt (shown in this jsbin) resulted in simultaneous transmission of all requests:
// JavaScript code snippet
// Define variables for request and response streams
var requestStream, responseStream;
// Populate the request stream with some sample data
requestStream = Rx.Observable.from(['a','b','c','d','e']);
// Use flatMap to handle the asynchronous nature of sendRequest function
responseStream = requestStream.flatMap(
sendRequest,
(val, response) => { return {val, response}; }
);
// Subscribe to the response stream to get the desired output
responseStream.subscribe(
item => {
console.log(item);
},
err => {
console.err(err);
},
() => {
console.log('Done');
}
);
// Function to handle request processing asynchronously
function sendRequest(val) {
return new Promise((resolve, reject) => {
setTimeout(() => { resolve('result for ' + val); }, 1000);
});
};
An alternative method that partially addresses the issue can be found in this jsbin, although it does not utilize a stream for request data:
// JavaScript code snippet
// Define variables for data and response stream
var data, responseStream;
data = ['a', 'b', 'c', 'd', 'e'];
// Create a custom observable to process the requests one by one
responseStream = Rx.Observable.create(observer => {
var sendNext = function() {
var val = data.shift();
if (!val) {
observer.onCompleted();
return;
}
sendRequest(val).then(response => {
observer.onNext({val, response});
sendNext();
});
};
sendNext();
});
// Subscribe to the response stream to observe the output
responseStream.subscribe(
item => {
console.log(item);
},
err => {
console.err(err);
},
() => {
console.log('Done');
}
);
// Function to simulate request handling with random response times
function sendRequest(val) {
return new Promise((resolve, reject) => {
setTimeout(() => { resolve('response for ' + val); }, Math.random() * 2500 + 500);
});
};
Your insights are greatly appreciated!
EDIT:
To clarify my objective, I aim to achieve the following sequence: "Send A, receive response for A, send B, receive response for B, send C, and so on..." Utilizing concatMap and defer as suggested by user3743222 seems to accomplish this task (check out this jsbin):
// JavaScript code snippet
// Implementing concatMap and defer for sequential processing
responseStream = requestStream.concatMap(
(val) => {
return Rx.Observable.defer(() => {
return sendRequest(val);
});
},
(val, response) => { return {val, response}; }
);