| 126 | } |
| 127 | |
| 128 | func (c *core) Start() error { |
| 129 | // increment the client reference counter |
| 130 | c.refCounter++ |
| 131 | |
| 132 | // Only start the core once |
| 133 | if c.started { |
| 134 | return nil |
| 135 | } |
| 136 | c.started = true |
| 137 | |
| 138 | c.closeChan = make(chan struct{}) |
| 139 | |
| 140 | // Connect to MessageBus, if we do not already have a connection |
| 141 | if c.mbus == nil { |
| 142 | switch messagebus.GetType(c.uri) { |
| 143 | case messagebus.TypeNats: |
| 144 | c.mbus = &messagebus.NatsBus{ |
| 145 | Config: messagebus.Config{ |
| 146 | URL: c.uri, |
| 147 | TimeoutRetries: c.timeoutRetries, |
| 148 | RequestTimeout: c.requestTimeout, |
| 149 | }, |
| 150 | Log: c.log, |
| 151 | } |
| 152 | case messagebus.TypeRabbitmq: |
| 153 | c.mbus = &messagebus.RabbitmqBus{ |
| 154 | Config: messagebus.Config{ |
| 155 | URL: "amqp://guest:guest@rabbitmq:5672/", |
| 156 | TimeoutRetries: c.timeoutRetries, |
| 157 | RequestTimeout: c.requestTimeout, |
| 158 | }, |
| 159 | Log: c.log, |
| 160 | } |
| 161 | default: |
| 162 | return errors.New("Unknown url for MessageBus: " + c.uri) |
| 163 | } |
| 164 | |
| 165 | err := c.mbus.Connect() |
| 166 | if err != nil { |
| 167 | c.close() |
| 168 | return eris.Wrap(err, "failed to connect to MessageBus") |
| 169 | } |
| 170 | c.closeMBusOnClose = true |
| 171 | } |
| 172 | |
| 173 | // Create and start the cluster |
| 174 | c.cluster = cluster.New() |
| 175 | |
| 176 | // Maintain the cluster |
| 177 | err := c.maintainCluster() |
| 178 | if err != nil { |
| 179 | c.close() |
| 180 | return eris.Wrap(err, "failed to start cluster maintenance") |
| 181 | } |
| 182 | |
| 183 | return nil |
| 184 | } |
| 185 | |