HTTP handler for incoming BLIP sync WebSocket request (/db/_blipsync)
()
| 29 | |
| 30 | // HTTP handler for incoming BLIP sync WebSocket request (/db/_blipsync) |
| 31 | func (h *handler) handleBLIPSync() error { |
| 32 | needRelease, err := h.server.incrementConcurrentReplications(h.rqCtx) |
| 33 | if err != nil { |
| 34 | h.db.DbStats.Database().NumReplicationsRejectedLimit.Add(1) |
| 35 | return err |
| 36 | } |
| 37 | // if we haven't incremented the active replicator due to MaxConcurrentReplications being 0, we don't need to decrement it |
| 38 | if needRelease { |
| 39 | defer h.server.decrementConcurrentReplications(h.rqCtx) |
| 40 | } |
| 41 | |
| 42 | // Exit early when the connection can't be switched to websocket protocol. |
| 43 | if !h.response.isHijackable() { |
| 44 | base.InfofCtx(h.ctx(), base.KeyHTTP, "Non-upgradable request received for BLIP+WebSocket protocol") |
| 45 | return base.HTTPErrorf(http.StatusUpgradeRequired, "Can't upgrade this request to websocket connection") |
| 46 | } |
| 47 | |
| 48 | h.db.DatabaseContext.DbStats.Database().NumReplicationsActive.Add(1) |
| 49 | h.db.DatabaseContext.DbStats.Database().NumReplicationsTotal.Add(1) |
| 50 | defer h.db.DatabaseContext.DbStats.Database().NumReplicationsActive.Add(-1) |
| 51 | |
| 52 | if c := h.server.Config.Replicator.BLIPCompression; c != nil { |
| 53 | blip.CompressionLevel = *c |
| 54 | } |
| 55 | |
| 56 | // error is checked at the time of database load, and ignored at this time |
| 57 | originPatterns, _ := hostOnlyCORS(h.db.CORS.Origin) |
| 58 | |
| 59 | cancelCtx, cancelCtxFunc := context.WithCancel(h.db.DatabaseContext.CancelContext) |
| 60 | // Create a BLIP context: |
| 61 | ctx, blipContext, err := db.NewSGBlipContext(h.ctx(), "", originPatterns, cancelCtx) |
| 62 | if err != nil { |
| 63 | cancelCtxFunc() |
| 64 | return err |
| 65 | } |
| 66 | |
| 67 | // Overwrite the existing logging context with the blip context ID |
| 68 | h.rqCtx = base.CorrelationIDLogCtx(h.ctx(), base.FormatBlipContextID(blipContext.ID)) |
| 69 | h.response.Header().Set(db.BLIPCorrelationIDResponseHeader, blipContext.ID) |
| 70 | // Create a new BlipSyncContext attached to the given blipContext. |
| 71 | bsc, err := db.NewBlipSyncContext(ctx, blipContext, h.db, db.BlipSyncStatsForCBL(h.db.DbStats), cancelCtxFunc) |
| 72 | if err != nil { |
| 73 | return err |
| 74 | } |
| 75 | defer bsc.Close() |
| 76 | |
| 77 | auditFields := base.AuditFields{base.AuditFieldReplicationID: base.FormatBlipContextID(blipContext.ID)} |
| 78 | if string(db.BLIPClientTypeSGR2) == h.getQuery(db.BLIPSyncClientTypeQueryParam) { |
| 79 | bsc.SetClientType(db.BLIPClientTypeSGR2) |
| 80 | auditFields["client_type"] = db.BLIPClientTypeSGR2 |
| 81 | } else { |
| 82 | // we could pull the exact CBL client and version from User-Agent |
| 83 | bsc.SetClientType(db.BLIPClientTypeCBL2) |
| 84 | auditFields["client_type"] = db.BLIPClientTypeCBL2 |
| 85 | } |
| 86 | base.Audit(h.rqCtx, base.AuditIDReplicationConnect, auditFields) |
| 87 | defer func() { |
| 88 | base.Audit(h.rqCtx, base.AuditIDReplicationDisconnect, auditFields) |
nothing calls this directly
no test coverage detected