lock/unlock
To enhance the functionality of the Deployer, consider incorporating a "lock" state -
// Deployer.js
class Deployer {
protected txDataList: Array<TXData>
protected isLocked: Boolean // lock state
constructor() {
this.txDataList = []
this.isLocked = false
setInterval(_ => this.read(), 5000)
}
async read() {
if (this.isLocked) return // exit if locked
this.isLocked = true // lock
let tx
while (true) {
tx = this.txDataList.shift()// fifo tx
if (tx == null) break // stop processing if no tx
await doSomething(tx) // process each tx
}
this.isLocked = false // unlock
}
write(tx: TXData) {
this.txDataList.push(tx)
}
}
export default Deployer
// some-service.js
import Deployer from "./Deployer.js"
const deployer = new Deployer()
someService.on("txData", tx => deployer.write(tx))
In this setup, we can write to the deployer at any time, but read only takes place when the deployer is in an "unlocked" state.
buffer
Rather than "lock/unlock," perhaps naming it Buffer would better represent the class behavior. Additionally, abstracting TXData allows us to use our Buffer for any data type. Let's also make the interval a configurable parameter -
// Buffer.js
class Buffer<T> { // generic T
private buffer: Array<T> // generic T
private isLocked: Boolean
constructor({ interval = 5 }) { // adjust interval as needed
this.buffer = []
this.isLocked = false
setInterval(_ => this.read(), interval * 1000)
}
async read() {
if (this.isLocked) return
this.isLocked = true
let t
while (true) {
t = this.buffer.shift()
if (t == null) break
await doSomething(t)
}
this.isLocked = false
}
write(t: T) { // generic T
this.buffer.push(t)
}
}
Type specification for Buffer<T> becomes Buffer<TXData>. Adjust the interval when creating a new instance of Buffer -
// some-service.js
import Buffer from "./Buffer.js"
const buffer = new Buffer<TXData>({ interval: 10 })
someService.on("txData", tx => buffer.write(tx))
move effect to call site
Let's remove the hard-coded doSomething function and provide an onRead parameter to allow the caller to define the action for each item -
// Buffer.js
class Buffer<T> {
private buffer: Array<T>
private isLocked: Boolean
private onRead: (x: T) => void // onRead
constructor({ interval = 5, onRead = console.log }) { // onRead
this.buffer = []
this.isLocked = false
setInterval(_ => this.read(), interval * 1000)
}
async read() {
if (this.isLocked) return
this.isLocked = true
let t
while (true) {
t = this.buffer.shift()
if (t == null) break
await this.onRead(t) // onRead
}
this.isLocked = false
}
write(t: T) {
this.buffer.push(t)
}
}
The caller now determines the effect previously handled by doSomething -
// some-service.js
import Buffer from "./Buffer.js"
async function doSomething(tx: TXData) {
await // process tx ...
}
const buffer = new Buffer<TXData>({
interval: 10,
onRead: doSomething // specify onRead
})
someService.on("txData", tx => buffer.write(tx))
related
For a related abstraction using async generators with a channel, check out this Q&A.