attak-processor/topology_utils.js at master · attak/attak-processor · GitHub

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
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
var TopologyUtils, nodePath;

nodePath = require('path');

TopologyUtils = {
  loadTopology: function(opts) {
    var environment, file, files, i, len, name, objDefs, procData, procName, processorPath, processors, ref, ref1, ref2, ref3, ref4, ref5, ref6, ref7, ref8, stream, streamName, streamsObj, topology, workingDir;
    workingDir = opts.cwd || process.cwd();
    environment = opts.environment || 'development';
    if (opts.topology) {
      if (opts.topology.constructor === String) {
        topology = require(nodePath.resolve(workingDir, opts.topology));
      } else {
        topology = opts.topology;
      }
    } else {
      topology = require(workingDir);
    }
    if (topology["default"]) {
      topology = topology["default"];
    }
    if (((ref = topology.streams) != null ? ref.constructor : void 0) === Function) {
      topology.streams = topology.streams();
    }
    if (((ref1 = topology.streams) != null ? ref1.constructor : void 0) === Array) {
      streamsObj = {};
      ref2 = topology.streams || {};
      for (streamName in ref2) {
        stream = ref2[streamName];
        if (stream.constructor === Array) {
          stream = {
            from: stream[0],
            to: stream[1],
            topic: stream[2]
          };
        }
        streamName = topology.name + "-" + stream.from + "-" + stream.to + "-" + environment;
        streamsObj[streamName] = stream;
      }
      topology.streams = streamsObj;
    }
    if (((ref3 = topology.processors) != null ? ref3.constructor : void 0) === Function) {
      topology.processors = topology.processors();
    } else if (((ref4 = topology.processors) != null ? ref4.constructor : void 0) === String) {
      processorPath = nodePath.resolve(workingDir, topology.processors);
      files = fs.readdirSync(processorPath);
      processors = {};
      for (i = 0, len = files.length; i < len; i++) {
        file = files[i];
        if (file === '.DS_Store') {
          continue;
        }
        name = nodePath.basename(file, nodePath.extname(file));
        processors[name] = topology.processors + "/" + file;
      }
      topology.processors = processors;
    } else if (topology.processors === void 0 && ((ref5 = topology.processor) != null ? ref5.constructor : void 0) === Function) {
      processors = {};
      ref6 = topology.streams;
      for (streamName in ref6) {
        stream = ref6[streamName];
        if (processors[stream.to] === void 0) {
          processors[stream.to] = stream.to;
        }
        if (processors[stream.from] === void 0) {
          processors[stream.from] = stream.from;
        }
      }
      topology.processors = processors;
    }
    ref7 = topology.processors || {};
    for (procName in ref7) {
      procData = ref7[procName];
      if (procData.constructor === String) {
        topology.processors[procName] = {
          path: procData
        };
      }
    }
    if (((ref8 = topology.api) != null ? ref8.constructor : void 0) === String) {
      objDefs = {
        handler: topology.api
      };
      topology.api = objDefs;
    }
    if (topology["static"]) {
      if (topology["static"].constructor === String) {
        topology["static"] = {
          "default": {
            dir: topology["static"]
          }
        };
      } else if (topology["static"].dir) {
        topology["static"] = {
          "default": topology["static"]
        };
      }
    }
    if (topology.auth) {
      if (topology.auth === true) {
        topology.auth = {
          "default": {}
        };
      }
    }
    return topology;
  },
  getProcessor: function(opts, topology, name) {
    var loading, procData, workingDir;
    workingDir = opts.cwd || process.cwd();
    if (topology.processor) {
      procData = topology.processor(name);
    } else if (topology.processors.constructor === String) {
      procData = topology.processors + "/" + name;
    } else if (topology.processors.constructor === Function) {
      procData = topology.processors()[name];
    } else {
      procData = topology.processors[name];
    }
    if (procData === void 0) {
      throw new Error("Failed to find processor " + name);
    }
    loading = {};
    if (procData.constructor === String) {
      loading = {
        type: 'path',
        path: procData
      };
    } else if ((procData != null ? procData.constructor : void 0) === Function) {
      loading = {
        type: 'dynamic',
        impl: procData
      };
    } else if ((procData != null ? procData.handler : void 0) === Function) {
      loading = {
        type: 'dynamic',
        impl: procData.handler
      };
    } else if ((procData != null ? procData.source : void 0) || (procData != null ? procData.path : void 0)) {
      loading = {
        type: 'path',
        path: procData.path
      };
    } else {
      loading = {
        type: 'dynamic',
        impl: procData
      };
    }
    switch (loading.type) {
      case 'path':
        loading.impl = opts.processor || require(nodePath.resolve(workingDir, loading.path));
    }
    return loading;
  }
};

module.exports = TopologyUtils;