// DispatchStreamHandler calls the stream handler, recovering from panics as
// errors.
//
// Deprecated: Use InvokeStreamHandler instead.
339938
func DispatchStreamHandler(
	h StreamHandler,
	stream *ServerStream,
) (err error) {
	return InvokeStreamHandler(StreamInvokeRequest{
		Stream:  stream,
		Handler: h,
	})
}
261378
func panicHandler(w http.ResponseWriter, r *http.Request, rcv interface{}) {
	ctx := r.Context()
	pe := &platform.Error{
		Code: platform.EInternal,
		Msg:  "a panic has occurred",
		Err:  fmt.Errorf("%v", rcv),
	}

	EncodeError(ctx, pe, w)
}
877023
func (h *BasicHost) newStreamHandler(s inet.Stream) {
	before := time.Now()

	if h.negtimeout > 0 {
		if err := s.SetDeadline(time.Now().Add(h.negtimeout)); err != nil {
			log.Error("setting stream deadline: ", err)
			s.Reset()
			return
		}
	}

	lzc, protoID, handle, err := h.Mux().NegotiateLazy(s)
	took := time.Now().Sub(before)
	if err != nil {
		if err == io.EOF {
			logf := log.Debugf
			if took > time.Second*10 {
				logf = log.Warningf
			}
			logf("protocol EOF: %s (took %s)", s.Conn().RemotePeer(), took)
		} else {
			log.Debugf("protocol mux failed: %s (took %s)", err, took)
		}
		s.Reset()
		return
	}

	s = &streamWrapper{
		Stream: s,
		rw:     lzc,
	}

	if h.negtimeout > 0 {
		if err := s.SetDeadline(time.Time{}); err != nil {
			log.Error("resetting stream deadline: ", err)
			s.Reset()
			return
		}
	}

	s.SetProtocol(protocol.ID(protoID))
	log.Debugf("protocol negotiation took %s", took)

	go handle(protoID, s)
}
1159247
func (s *Session) handleStreamMessage(hdr header) error {
	// Check for a new stream creation
	id := hdr.StreamID()
	flags := hdr.Flags()
	if flags&flagSYN == flagSYN {
		if err := s.incomingStream(id); err != nil {
			return err
		}
	}

	// Get the stream
	s.streamLock.Lock()
	stream := s.streams[id]
	s.streamLock.Unlock()

	// If we do not have a stream, likely we sent a RST
	if stream == nil {
		// Drain any data on the wire
		if hdr.MsgType() == typeData && hdr.Length() > 0 {
			s.logger.Printf("[WARN] yamux: Discarding data for stream: %d", id)
			if _, err := io.CopyN(ioutil.Discard, s.bufRead, int64(hdr.Length())); err != nil {
				s.logger.Printf("[ERR] yamux: Failed to discard data: %v", err)
				return nil
			}
		} else {
			s.logger.Printf("[WARN] yamux: frame for missing stream: %v", hdr)
		}
		return nil
	}

	// Check if this is a window update
	if hdr.MsgType() == typeWindowUpdate {
		if err := stream.incrSendWindow(hdr, flags); err != nil {
			if sendErr := s.sendNoWait(s.goAway(goAwayProtoErr)); sendErr != nil {
				s.logger.Printf("[WARN] yamux: failed to send go away: %v", sendErr)
			}
			return err
		}
		return nil
	}

	// Read the new data
	if err := stream.readData(hdr, flags, s.bufRead); err != nil {
		if sendErr := s.sendNoWait(s.goAway(goAwayProtoErr)); sendErr != nil {
			s.logger.Printf("[WARN] yamux: failed to send go away: %v", sendErr)
		}
		return err
	}
	return nil
}
493419
func (self *FSock) dispatchEvent(event string) {
	eventName := headerVal(event, "Event-Name")
	if eventName == "BACKGROUND_JOB" { // for bgapi BACKGROUND_JOB
		go self.doBackgroudJob(event)
		return
	}

	if eventName == "CUSTOM" {
		eventSubclass := headerVal(event, "Event-Subclass")
		if len(eventSubclass) != 0 {
			eventName += " " + urlDecode(eventSubclass)
		}
	}

	for _, handleName := range []string{eventName, "ALL"} {
		if _, hasHandlers := self.eventHandlers[handleName]; hasHandlers {
			// We have handlers, dispatch to all of them
			for _, handlerFunc := range self.eventHandlers[handleName] {
				go handlerFunc(event, self.connId)
			}
			return
		}
	}
	if self.logger != nil {
		fmt.Printf("No dispatcher, event name: %s, handlers: %+v\n", eventName, self.eventHandlers)
		self.logger.Warning(fmt.Sprintf("<FSock> No dispatcher for event: <%+v>", event))
	}
}
802451
func (ed *Dispatcher) RegisterHandler(t interface{}, h Handler) {
	htype := reflect.TypeOf(t)
	if _, ok := ed.handlers[htype]; !ok {
		logger.Debugf("Registering handler for %s on dispatcher %T", htype, ed)
		ed.handlers[htype] = h
	} else {
		logger.Debugf("Cannot register handler %s on dispatcher %T since it's already registered", htype, ed)
	}
}
314348
func NewStreamWatcher(d Decoder) *StreamWatcher {
	sw := &StreamWatcher{
		source: d,
		// It's easy for a consumer to add buffering via an extra
		// goroutine/channel, but impossible for them to remove it,
		// so nonbuffered is better.
		result: make(chan Event),
	}
	go sw.receive()
	return sw
}
643823
func NewStreamHandler(
	name string, level LogLevelType, stream Stream) *StreamHandler {

	object := &StreamHandler{
		BaseHandler: NewBaseHandler(name, level),
		stream:      stream,
	}
	Closer.AddHandler(object)
	return object
}
459279
func (s *Swarm) SetStreamHandler(handler inet.StreamHandler) {
	s.streamh.Store(handler)
}
1083127
func (d *eventDispatcher) DispatchEvent(e Event) {
	d.RLock()
	defer d.RUnlock()

	// Automatically set the event source.
	if e, ok := e.(*event); ok {
		e.source = d.source
	}

	// Dispatch the event to all listeners.
	for _, l := range d.listeners[e.Type()] {
		l(e)
	}
}
1287838
func (r *Router) PanicHandler(panicHandler func(*Context, interface{})) {
	r.initRouter()
	r.router.PanicHandler = panicHandler
}
151144
func PanicFlatHandler(fx FlatHandler, p PanicHandler) FlatHandler {
	return func(c *Context, next NextHandler) {
		defer func() {
			if err := recover(); err != nil {
				p(c.Res, c.Req, err)
			}
		}()

		fx(c, next)
	}
}
1311227
func NewStreamWatcher(d Decoder, r Reporter) *StreamWatcher {
	sw := &StreamWatcher{
		source:   d,
		reporter: r,
		// It's easy for a consumer to add buffering via an extra
		// goroutine/channel, but impossible for them to remove it,
		// so nonbuffered is better.
		result: make(chan Event),
	}
	go sw.receive()
	return sw
}
352710
func HeaderHandler(headers ...string) func(*http.Request) map[string]interface{} {
	return func(req *http.Request) map[string]interface{} {
		result := map[string]interface{}{}
		for _, header := range headers {
			if val := req.Header.Get(header); val != "" {
				result[header] = val
			}
		}
		return result
	}
}
816665
func DispatchUnaryHandler(
	ctx context.Context,
	h UnaryHandler,
	start time.Time,
	req *Request,
	resq ResponseWriter,
) (err error) {
	return InvokeUnaryHandler(UnaryInvokeRequest{
		Context:        ctx,
		StartTime:      start,
		Request:        req,
		ResponseWriter: resq,
		Handler:        h,
	})
}
870345
func StreamHandler(w http.ResponseWriter, r *http.Request) {
	n, _ := strconv.Atoi(mux.Vars(r)["n"]) // shouldn't fail due to route pattern
	nl := []byte{'\n'}
	// allow only millisecond precision
	for i := 0; i < n; i++ {
		time.Sleep(StreamInterval)
		b, _ := json.Marshal(struct {
			N    int       `json:"n"`
			Time time.Time `json:"time"`
		}{i, time.Now().UTC()})
		w.Write(b)
		w.Write(nl)
		if f, ok := w.(http.Flusher); ok {
			f.Flush()
		}
	}
}
1307078
func (self *RawDispatcher) Dispatch(mux *ServeMux, w http.ResponseWriter, r *http.Request) *ServeMux {
	//check the prefix for sanity
	pre := self.Prefix + "/"
	path := r.URL.Path
	if strings.HasSuffix(path, "/") && path != "/" {
		path = path[0 : len(path)-1]
	}
	if self.Prefix != "" {
		if !strings.HasPrefix(path, pre) {
			panic(fmt.Sprintf("expected prefix %s on the URL path but not found on: %s", pre, path))
		}
		path = path[len(pre):]
	}
	parts := strings.Split(path, "/")
	bundle, err := self.IO.BundleHook(w, r, self.SessionMgr)
	if err != nil {
		http.Error(w, fmt.Sprintf("failed to create parameter bundle:%s", err), http.StatusInternalServerError)
		return nil
	}
	self.DispatchSegment(mux, w, r, parts, self.Root, bundle)
	return nil
}