Method beginStreams

reb/globrun.go:437–453

(config *cmn.Config)

}

func (reb *Reb) beginStreams(config *cmn.Config) {
	debug.Assert(reb.stages.stage.Load() == rebStageInit)

	xreb := reb.xctn()
	reb.dm.SetXact(xreb)
	reb.dm.Open()
	pushArgs := bundle.Args{
		Net:        reb.dm.NetC(),
		Trname:     trnamePsh,
		Multiplier: config.Rebalance.SbundleMult,
		Extra:      &transport.Extra{SenderID: xreb.ID()},
	}
	reb.pushes = bundle.NewStreams(reb.t.Sowner(), reb.t.Snode(), transport.NewIntraDataClient(), pushArgs)

	reb.laterx.Store(false)
	reb.inQueue.Store(0)
}

func (reb *Reb) abortStreams() {
	reb.dm.Abort()