forked from falconair/nodefix
-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathFIXServer.js
129 lines (106 loc) · 3.89 KB
/
FIXServer.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
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
const { Observable, Subject } = require('rx')
var net = require('net');
var fixutil = require('./fixutils.js');
var {FrameDecoder} = require('./handlers/FrameDecoder')
var {FIXSession} = require('./handlers/FIXSession')
exports.fixutil = fixutil;
exports.FIXServer = function(opt) {
var self = this
var HOST
var PORT
this.fixSessions = {}
this.connect$ = new Subject
this.logon$ = new Subject
this.logoff$ = new Subject
this.fixIn$ = new Subject
this.dataIn$ = new Subject
this.jsonIn$ = new Subject
this.fixOut$ = new Subject
this.dataOut$ = new Subject
this.jsonOut$ = new Subject
this.end$ = new Subject
this.close$ = new Subject
this.error$ = new Subject
var server = net.createServer(function(connection) {
var sessionHolder = {}
const frameDecoder = new FrameDecoder()
const fixSession = new FIXSession(sessionHolder, true, opt)
var senderId;
sessionHolder.connection = connection
var logon$ = Observable.fromEvent(fixSession, 'logon')
logon$.subscribe(self.logon$)
var logoff$ = Observable.fromEvent(fixSession, 'logoff')
logoff$.subscribe(self.logoff$)
logoff$.subscribe((x)=>{ delete self.fixSessions[senderId] })
var dataOut$ = Observable.fromEvent(fixSession, 'dataOut')
dataOut$.subscribe(self.dataOut$)
var fixOut$ = Observable.fromEvent(fixSession, 'fixOut')
fixOut$.subscribe(self.fixOut$)
var jsonOut$ = fixOut$
.map((msg) => {
return {msg: fixutil.convertToJSON(msg), senderId: senderId}
})
jsonOut$.subscribe(self.jsonOut$)
var end$ = Observable.fromEvent(connection, 'end')
.map((x)=>{ return senderId })
end$.subscribe(self.end$)
end$.subscribe((x)=>{ delete self.fixSessions[x] })
var close$ = Observable.fromEvent(connection, 'close')
.map((x)=>{ return senderId })
close$.subscribe(self.close$)
close$.subscribe((x)=>{ delete self.fixSessions[x] })
var error$ = Observable.fromEvent(connection, 'error')
.map((x)=>{ return { error: x, senderId: senderId }})
error$.subscribe(self.error$)
var rawIn$ = Observable.fromEvent(connection, 'data')
var fixIn$ = rawIn$
.flatMap((raw) => { return frameDecoder.decode(raw)})
fixIn$.subscribe(self.fixIn$)
var jsonIn$ = fixIn$
.map((msg) => {
return {msg: fixutil.convertToJSON(msg), senderId: senderId}
})
.catch((ex)=>{
self.connection.emit('error', ex)
return Observable.never()}
)
.share()
jsonIn$.subscribe(self.jsonIn$)
var dataIn$ = fixIn$
.map((msg) => {
return {msg: fixSession.decode(msg), senderId: senderId}
})
.catch((ex)=>{
connection.emit('error', ex)
return Observable.never()}
)
.share()
dataIn$.subscribe(self.dataIn$)
logon$.subscribe((x) => {
senderId = x
self.fixSessions[x] = sessionHolder
})
sessionHolder.send = function(fix) {
fixSession.send(fix)
}
sessionHolder.resetFIXSession = function(){
fixSession.resetFIXSession()
}
})
this.listen = function(port, host, callback) {
PORT = port
HOST = host
server.listen(port, host, callback);
}
this.send = function(targrtId, fix) {
const sessionHolder = server.fixSessions[targrtId]
if(sessionHolder)
sessionHolder.send(fix)
}
this.resetFIXSession = function(targrtId){
const sessionHolder = server.fixSessions[targrtId]
if(sessionHolder)
sessionHolder.resetFIXSession()
}
return this
}