All files / Nodejs/connectors/dw toDW.js

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
"use strict";
 
var extend = require("extend");
var refUtil = require("../../lib/reference.js");
var moment = require("moment");
 
module.exports = function (configure) {
	configure = configure || {};
	let leo = require("../../index")(configure);
	let ls = leo.streams;
 
	return {
		validate: function () {},
		stream: function (suffix) {
			let eventName = "queue:dw.load" + (suffix ? suffix : "");
			return ls.through((obj, done) => {
				if (!obj.payload){
					obj = {payload:obj};
				}
				obj.event = eventName;
				done(null, obj);
			});
		},
		write: function (id, suffix) {
			return ls.pipeline(this.stream(suffix), ls.through((event, done)=>{
				event.id = id;
				done(null, event)
			}), leo.write(id, {
				firehose: true,
				debug: true
			}), ls.toCheckpoint({
				debug: true
			}));
		},
		run: function (id, source, transform, opts, callback) {
			if (typeof opts === "function") {
				callback = opts;
				opts = {};
			}
 
			opts = Object.assign({
				debug: false
			}, opts);
 
			return ls.pipe(
				leo.read(id, source, {
					debug: opts.debug
				}),
				ls.process(id, transform),
				this.write(id, opts.suffix),
				callback
			);
		}
	}
};