NewInternalAPI returns a concerete implementation of the internal API. Callers can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes.
( base *base.BaseDendrite, federation api.FederationClient, rsAPI roomserverAPI.FederationRoomserverAPI, caches *caching.Caches, keyRing *gomatrixserverlib.KeyRing, resetBlacklist bool, )
| 95 | // NewInternalAPI returns a concerete implementation of the internal API. Callers |
| 96 | // can call functions directly on the returned API or via an HTTP interface using AddInternalRoutes. |
| 97 | func NewInternalAPI( |
| 98 | base *base.BaseDendrite, |
| 99 | federation api.FederationClient, |
| 100 | rsAPI roomserverAPI.FederationRoomserverAPI, |
| 101 | caches *caching.Caches, |
| 102 | keyRing *gomatrixserverlib.KeyRing, |
| 103 | resetBlacklist bool, |
| 104 | ) api.FederationInternalAPI { |
| 105 | cfg := &base.Cfg.FederationAPI |
| 106 | |
| 107 | federationDB, err := storage.NewDatabase(base, &cfg.Database, base.Caches, base.Cfg.Global.ServerName) |
| 108 | if err != nil { |
| 109 | logrus.WithError(err).Panic("failed to connect to federation sender db") |
| 110 | } |
| 111 | |
| 112 | if resetBlacklist { |
| 113 | _ = federationDB.RemoveAllServersFromBlacklist() |
| 114 | } |
| 115 | |
| 116 | stats := &statistics.Statistics{ |
| 117 | DB: federationDB, |
| 118 | FailuresUntilBlacklist: cfg.FederationMaxRetries, |
| 119 | } |
| 120 | |
| 121 | js, _ := base.NATS.Prepare(base.ProcessContext, &cfg.Matrix.JetStream) |
| 122 | |
| 123 | queues := queue.NewOutgoingQueues( |
| 124 | federationDB, base.ProcessContext, |
| 125 | cfg.Matrix.DisableFederation, |
| 126 | cfg.Matrix.ServerName, federation, rsAPI, stats, |
| 127 | &queue.SigningInfo{ |
| 128 | KeyID: cfg.Matrix.KeyID, |
| 129 | PrivateKey: cfg.Matrix.PrivateKey, |
| 130 | ServerName: cfg.Matrix.ServerName, |
| 131 | }, |
| 132 | ) |
| 133 | |
| 134 | rsConsumer := consumers.NewOutputRoomEventConsumer( |
| 135 | base.ProcessContext, cfg, js, queues, |
| 136 | federationDB, rsAPI, |
| 137 | ) |
| 138 | if err = rsConsumer.Start(); err != nil { |
| 139 | logrus.WithError(err).Panic("failed to start room server consumer") |
| 140 | } |
| 141 | tsConsumer := consumers.NewOutputSendToDeviceConsumer( |
| 142 | base.ProcessContext, cfg, js, queues, federationDB, |
| 143 | ) |
| 144 | if err = tsConsumer.Start(); err != nil { |
| 145 | logrus.WithError(err).Panic("failed to start send-to-device consumer") |
| 146 | } |
| 147 | receiptConsumer := consumers.NewOutputReceiptConsumer( |
| 148 | base.ProcessContext, cfg, js, queues, federationDB, |
| 149 | ) |
| 150 | if err = receiptConsumer.Start(); err != nil { |
| 151 | logrus.WithError(err).Panic("failed to start receipt consumer") |
| 152 | } |
| 153 | typingConsumer := consumers.NewOutputTypingConsumer( |
| 154 | base.ProcessContext, cfg, js, queues, federationDB, |