code: func (r *Registry) Subscribe(peerId enode.ID, s Stream, h *Range, priority uint8) error {
  	// check if the stream is registered
  	if _, err := r.GetClientFunc(s.Name); err != nil {
  		return err
  	}
  
  	peer := r.getPeer(peerId)
  	if peer == nil {
  		return fmt.Errorf("peer not found %v", peerId)
  	}
  
  	var to uint64
  	if !s.Live && h != nil {
  		to = h.To
  	}
  
  	err := peer.setClientParams(s, newClientParams(priority, to))
  	if err != nil {
  		return err
  	}
  	if s.Live && h != nil {
  		if err := peer.setClientParams(
  			getHistoryStream(s),
  			newClientParams(getHistoryPriority(priority), h.To),
  		); err != nil {
  			return err
  		}
  	}
  
  	msg := &SubscribeMsg{
  		Stream:   s,
  		History:  h,
  		Priority: priority,
  	}
  	log.Debug("Subscribe ", "peer", peerId, "stream", s, "history", h)
  
  	return peer.SendPriority(context.TODO(), msg, priority)
  }
docs: // Subscribe initiates the streamer