}
      }
    }
  }

  /**
   * Returns the number of [open streams][Http2Stream.isOpen] on this connection.
   */
  fun openStreamCount(): Int = withLock { streams.size }

  fun getStream(id: Int): Http2Stream? = withLock { streams[id] }

  internal fun removeStream(streamId: Int): Http2Stream? {
    withLock {
      val stream = streams.remove(streamId)

			target.loggerOnce(context.Background(), err, target.ID().String())
		}
		return err
	}

	if target.natsConn != nil && args.JetStream.Enable {
		var jstream nats.JetStream
		jstream, err = target.natsConn.JetStream()
		if err != nil {
			if err.Error() != nats.ErrNoServers.Error() {
				target.loggerOnce(context.Background(), err, target.ID().String())
			}
			return err
		}

			PingInterval:      pingInterval,
			QueueDir:          env.Get(queueDirEnv, kv.Get(target.NATSQueueDir)),
			QueueLimit:        queueLimit,
			RootCAs:           rootCAs,
		}
		natsArgs.JetStream.Enable = env.Get(jetStreamEnableEnv, kv.Get(target.NATSJetStream)) == config.EnableOn

		streamingEnableEnv := target.EnvNATSStreaming
		if k != config.Default {

* Add external link [Apache Kafka producer and consumer with FastAPI and aiokafka](https://iwpnd.pw/articles/2020-03/apache-kafka-fastapi-geostream) to docs. PR [#1112](https://github.com/tiangolo/fastapi/pull/1112) by [@iwpnd](https://github.com/iwpnd).