Method runForwardSubscriber

worker/import.go:154–215

(ctx context.Context, out api.Dgraph_StreamExtSnapshotClient, peerId string)

}

func (ps *pubSub) runForwardSubscriber(ctx context.Context, out api.Dgraph_StreamExtSnapshotClient, peerId string) error {
	defer func() {
		glog.Infof("[import] forward subscriber stopped for peer [%v]", peerId)
	}()

	buffer := ps.subscribe()
	defer ps.unsubscribe(buffer) // ensure publisher won't block on us if we exit

Loop:
	for {
		select {
		case <-ctx.Done():
			glog.Infof("[import] Context cancelled, stopping receive goroutine: %v", ctx.Err())
			return ctx.Err()

		default:
			msg, ok := <-buffer
			if !ok {
				break Loop
			}

			if msg.Pkt.Done {
				glog.Infof("[import] received done signal from [%v]", peerId)
				d := api.StreamPacket{Done: true}
				if err := out.Send(&api.StreamExtSnapshotRequest{Pkt: &d}); err != nil {
					return err
				}

				_ = out.CloseSend()

				for {
					if ctx.Err() != nil {
						return ctx.Err()
					}
					r, err := out.Recv()
					if errors.Is(err, io.EOF) {
						return fmt.Errorf("server closed stream before Finish=true for peer [%v]", peerId)
					}
					if err != nil {
						return fmt.Errorf("failed to receive final response from peer [%v]: %w", peerId, err)
					}
					if r.Finish {
						glog.Infof("[import] peer [%v]: Received final Finish=true", peerId)
						break Loop
					}
					glog.Infof("[import] peer [%v]: Waiting for Finish=true, got interim ACK", peerId)
				}
			}

			data := &api.StreamExtSnapshotRequest{Pkt: &api.StreamPacket{Data: msg.Pkt.Data}}
			if err := out.Send(data); err != nil {
				return err
			}

			if _, err := out.Recv(); err != nil {
				return fmt.Errorf("failed to receive response from peer [%v]: %w", peerId, err)
			}
		}