Make sure NATS uses own persistent directory (TODO: make this configurable)

setup/config/config.go

@@ -329,6 +329,7 @@ func (c *Dendrite) Verify(configErrs *ConfigErrors, isMonolith bool) {
 }
 
 func (c *Dendrite) Wiring() {
+	c.Global.Kafka.Matrix = &c.Global
 	c.ClientAPI.Matrix = &c.Global
 	c.EDUServer.Matrix = &c.Global
 	c.FederationAPI.Matrix = &c.Global

setup/config/config_kafka.go

@@ -23,6 +23,8 @@ var KafkaTopics = []string{
 }
 
 type Kafka struct {
+	Matrix *Global `yaml:"-"`
+	
 	// A list of kafka/NATS addresses to connect to.
 	Addresses []string `yaml:"addresses"`
 	// The prefix to use for Kafka topic names for this homeserver - really only

setup/kafka/kafka.go

@@ -37,6 +37,7 @@ func SetupConsumerProducer(cfg *config.Kafka) (sarama.Consumer, sarama.SyncProdu
 					ServerName: "monolith",
 					DontListen: true,
 					JetStream: true,
+					StoreDir: string(cfg.Matrix.ServerName),
 					LogFile: "nats.log",
 					Debug: true,
 				})
@@ -48,7 +49,9 @@ func SetupConsumerProducer(cfg *config.Kafka) (sarama.Consumer, sarama.SyncProdu
 				s = natsServer
 			}
 			natsServerMutex.Unlock()
-			natsServer.WaitForStartup()
+			if !natsServer.ReadyForConnections(time.Second * 10) {
+				logrus.Fatalln("NATS did not start in time")
+			}
 			conn, err := s.InProcessConn()
 			if err != nil {
 				logrus.Fatalln("Failed to get a NATS in-process conn")