connect establishes a blip connection to a remote host.
(arc *activeReplicatorCommon, idSuffix string)
| 208 | |
| 209 | // connect establishes a blip connection to a remote host. |
| 210 | func connect(arc *activeReplicatorCommon, idSuffix string) (blipSender *blip.Sender, bsc *BlipSyncContext, err error) { |
| 211 | arc.replicationStats.NumConnectAttempts.Add(1) |
| 212 | |
| 213 | ctx := base.CorrelationIDLogCtx( |
| 214 | arc.config.ActiveDB.AddDatabaseLogContext(base.NewNonCancelCtx().Ctx), |
| 215 | arc.config.ID+idSuffix) |
| 216 | if arc.config.RunAs != "" { |
| 217 | ctx = base.UserLogCtx(ctx, arc.config.RunAs, base.UserDomainSyncGateway, nil) |
| 218 | } else { |
| 219 | ctx = arc.config.ActiveDB.AddBucketUserLogContext(ctx) |
| 220 | } |
| 221 | |
| 222 | cancelCtx, cancelFunc := context.WithCancel(context.WithoutCancel(ctx)) // separate cancel context from parent cancel context |
| 223 | |
| 224 | var originPatterns []string // no origin headers for ISGR |
| 225 | var blipContext *blip.Context |
| 226 | |
| 227 | // NewSGBlipContext doesn't set cancellation context - active replication cancellation on db close is handled independently |
| 228 | if len(arc.config.SupportedBLIPProtocols) > 0 { |
| 229 | ctx, blipContext, err = NewSGBlipContextWithProtocols(ctx, arc.config.ID+idSuffix, originPatterns, arc.config.SupportedBLIPProtocols, cancelCtx) |
| 230 | } else { |
| 231 | ctx, blipContext, err = NewSGBlipContext(ctx, arc.config.ID+idSuffix, originPatterns, cancelCtx) |
| 232 | } |
| 233 | if err != nil { |
| 234 | cancelFunc() |
| 235 | return nil, nil, err |
| 236 | } |
| 237 | blipContext.WebsocketPingInterval = arc.config.WebsocketPingInterval |
| 238 | blipContext.OnExitCallback = func() { |
| 239 | // fall into a reconnect loop only if the connection is unexpectedly closed. |
| 240 | if ctx.Err() == nil { |
| 241 | arc.reconnect() |
| 242 | } |
| 243 | } |
| 244 | |
| 245 | bsc, err = NewBlipSyncContext(ctx, blipContext, arc.config.ActiveDB, arc.replicationStats, cancelFunc) |
| 246 | if err != nil { |
| 247 | return nil, nil, err |
| 248 | } |
| 249 | |
| 250 | // NewBlipSyncContext has already set deltas as disabled/enabled based on config.ActiveDB. |
| 251 | // If deltas have been disabled in the replication config, override this value |
| 252 | if arc.config.DeltasEnabled == false { |
| 253 | bsc.sgCanUseDeltas = false |
| 254 | } |
| 255 | |
| 256 | blipSender, err = blipSync(*arc.config.RemoteDBURL, blipContext, arc.config.InsecureSkipVerify) |
| 257 | if err != nil { |
| 258 | return nil, nil, err |
| 259 | } |
| 260 | |
| 261 | // set client type to SGW on active peer |
| 262 | bsc.SetClientType(BLIPClientTypeSGR2) |
| 263 | |
| 264 | // set active subprotocol after handshake |
| 265 | err = bsc.SetActiveCBMobileSubprotocol(blipContext.ActiveSubprotocol()) |
| 266 | if err != nil { |
| 267 | return nil, nil, err |
no test coverage detected