-
Notifications
You must be signed in to change notification settings - Fork 1
/
exchange.go
157 lines (139 loc) · 4.15 KB
/
exchange.go
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
package main
import (
"crypto/tls"
"encoding/json"
"io/ioutil"
"net/http"
"time"
)
// Parser interface implemented by concreate exchanges
type Parser interface {
Parse(meta Exchange, row map[string]interface{}) TradeRow
}
// -- Exchange struct ---
type Exchange struct {
Name string
Pair string
Url string
Parser Parser
Interval int
}
// Gets the data from the given URL and parse it using JSON parser
func (exchange *Exchange) Fetch() (data []map[string]interface{}, err error) {
resp, err := EXCHANGES.HttpClient.Get(exchange.Url)
if err != nil {
LOGGER.Error.Println("Error occurred fetching data from exchange [", exchange.Name, "]: ", err)
return nil, err
}
defer resp.Body.Close()
var resBody, resErr = ioutil.ReadAll(resp.Body)
if resErr != nil {
LOGGER.Error.Println("Error occurred reading response body [", exchange.Name, "]: ", resErr)
return nil, resErr
}
err = json.Unmarshal(resBody, &data)
if err != nil {
LOGGER.Error.Println("Error occurred Parsing data->json from exchange [", exchange.Name, "]: ", err)
return nil, err
}
return data, nil
}
// Convert raw data into TradeRow struct
func (exchange *Exchange) Parse(rows []map[string]interface{}) (tradeRows []TradeRow, err error) {
// During parsing, we could encounter panic due to data mismatch or json errors.
defer func() {
if r := recover(); r != nil {
LOGGER.Error.Println("Error occured in parsing on exchange [", exchange.Name, "]: ", r)
tradeRows = []TradeRow{}
err = nil // we ignore parsing errors and moving on
}
}()
tradeRows = []TradeRow{}
for _, row := range rows {
tradeRow := exchange.Parser.Parse(*exchange, row)
tradeRows = append(tradeRows, tradeRow)
}
return tradeRows, nil
}
// -- End of Exchange --- //
// -- Exchanges struct --- //
type Exchanges struct {
HttpClient *http.Client
_Exchanges []Exchange
}
// Register the http client
func (exchanges *Exchanges) RegisterHttpClient(httpClient *http.Client) {
exchanges.HttpClient = httpClient
}
// Register an exchange
func (exchanges *Exchanges) RegisterExchange(exchange Exchange) {
exchanges._Exchanges = append(exchanges._Exchanges, exchange)
}
func (exchanges *Exchanges) GetExchanges() []Exchange {
return exchanges._Exchanges
}
func (exchanges *Exchanges) ClearExchanges() {
exchanges._Exchanges = []Exchange{}
}
// Listen function launches registered exchanges as go-routines
func (exchanges *Exchanges) Listen(closeChannel chan bool, broadcastChannel chan []TradeRow) {
for _, _exchange := range exchanges._Exchanges {
// Each exchange gets its own go-routine
go func(exchange Exchange) {
LOGGER.Trace.Println("Listening for exchange: ", exchange.Name)
// Each exchange pulls data on a specific interval
var interval = time.Duration(exchange.Interval) * time.Second
ticker := time.NewTicker(interval)
// Close out the ticker timer on the exit
// TIP: Nice way of using defer functions to close out any resources before go-routine exits
// We could have done in closechannel select as well but this seems like more natural
defer func() {
ticker.Stop()
}()
// Infinite loop until instructed to close
// Fetch on each ticker and broadcast the data on the given channel
for {
select {
case <-ticker.C:
if broadcastChannel != nil {
var data, err = exchange.Fetch()
if err != nil {
continue
}
var processedData, _ = exchange.Parse(data)
broadcastChannel <- processedData
}
case <-closeChannel:
broadcastChannel = nil
return
}
}
}(_exchange)
}
}
var EXCHANGES = Exchanges{}
// -- End of Exchanges --- //
// -- Traderow Struct to hold Trade data --//
type TradeRow struct {
Triplet string
Tid string
Timestamp string
Amount float64
Price float64
}
// Convert to JSON string
func (tradeRow *TradeRow) ToJson() []byte {
var tradeRowAsJson, err = json.Marshal(tradeRow)
if err != nil {
//TODO: raise error
}
return tradeRowAsJson
}
//-- End of Traderow ---//
// Setup HTTP Client to accept all HTTPS domains
func init() {
tr := &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
}
EXCHANGES.RegisterHttpClient(&http.Client{Transport: tr})
}