Function getTopicConsumeSSEHandler

_examples/kafka-api/main.go:304–391

(ctx iris.Context)

}

func getTopicConsumeSSEHandler(ctx iris.Context) {
	flusher, ok := ctx.ResponseWriter().Flusher()
	if !ok {
		ctx.StopWithText(iris.StatusHTTPVersionNotSupported, "streaming unsupported")
		return
	}

	ctx.ContentType("application/json, text/event-stream")
	ctx.Header("Cache-Control", "no-cache")
	ctx.Header("Connection", "keep-alive")

	master, err := sarama.NewConsumer(brokers, config)
	if err != nil {
		fail(ctx, iris.StatusInternalServerError, "unable to start master consumer: %v", err)
		return
	}

	fromTopic := ctx.Params().Get("topic")
	// take the partition, defaults to the first found if not url query parameter "partition" passed.
	var partition int32
	partitions, err := master.Partitions(fromTopic)
	if err != nil {
		master.Close()
		fail(ctx, iris.StatusInternalServerError, "unable to get partitions for topic: '%s': %v", fromTopic, err)
		return
	}

	if len(partitions) > 0 {
		partition = partitions[0]
	}

	partition = ctx.URLParamInt32Default("partition", partition)
	offset := ctx.URLParamInt64Default("offset", sarama.OffsetOldest)

	consumer, err := master.ConsumePartition(fromTopic, partition, offset)
	if err != nil {
		ctx.Application().Logger().Error(err)
		master.Close() // close the master here to avoid any leaks, we will exit.
		fail(ctx, iris.StatusInternalServerError, "unable to start partition consumer: %v", err)
		return
	}

	// `OnClose` fires when the request is finally done (all data read and handler exits) or interrupted by the user.
	ctx.OnClose(func(_ iris.Context) {
		ctx.Application().Logger().Warnf("a client left")

		// Close shuts down the consumer. It must be called after all child
		// PartitionConsumers have already been closed. <-- That is what
		// godocs says but it doesn't work like this.
		// if err = consumer.Close(); err != nil {
		// 	ctx.Application().Logger().Errorf("[%s] unable to close partition consumer: %v", ctx.RemoteAddr(), err)
		// }
		// so close the master only and omit the first ^ consumer.Close:
		if err = master.Close(); err != nil {
			ctx.Application().Logger().Errorf("[%s] unable to close master consumer: %v", ctx.RemoteAddr(), err)
		}
	})