mirror of
https://github.com/matrix-org/dendrite.git
synced 2025-12-06 14:33:10 -06:00
Accept a list of kafka producer URIs
This commit is contained in:
parent
49ed708ca4
commit
3a78a0b8b0
|
|
@ -12,6 +12,7 @@ import (
|
||||||
|
|
||||||
log "github.com/Sirupsen/logrus"
|
log "github.com/Sirupsen/logrus"
|
||||||
"github.com/matrix-org/dugong"
|
"github.com/matrix-org/dugong"
|
||||||
|
sarama "gopkg.in/Shopify/sarama.v1"
|
||||||
)
|
)
|
||||||
|
|
||||||
func setupLogging(logDir string) {
|
func setupLogging(logDir string) {
|
||||||
|
|
@ -38,17 +39,28 @@ func main() {
|
||||||
if logDir != "" {
|
if logDir != "" {
|
||||||
setupLogging(logDir)
|
setupLogging(logDir)
|
||||||
}
|
}
|
||||||
log.Info("Starting clientapi")
|
|
||||||
// TODO: Rather than generating a new key on every startup, we should be
|
// TODO: Rather than generating a new key on every startup, we should be
|
||||||
// reading a PEM formatted file instead.
|
// reading a PEM formatted file instead.
|
||||||
_, privKey, err := ed25519.GenerateKey(nil)
|
_, privKey, err := ed25519.GenerateKey(nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Panicf("Failed to generate private key: %s", err)
|
log.Panicf("Failed to generate private key: %s", err)
|
||||||
}
|
}
|
||||||
routing.Setup(http.DefaultServeMux, http.DefaultClient, config.ClientAPI{
|
|
||||||
ServerName: "localhost",
|
cfg := config.ClientAPI{
|
||||||
KeyID: "ed25519:something",
|
ServerName: "localhost",
|
||||||
PrivateKey: privKey,
|
KeyID: "ed25519:something",
|
||||||
})
|
PrivateKey: privKey,
|
||||||
|
KafkaProducerURIs: []string{"localhost"},
|
||||||
|
}
|
||||||
|
|
||||||
|
log.Info("Starting clientapi")
|
||||||
|
|
||||||
|
producer, err := sarama.NewSyncProducer(cfg.KafkaProducerURIs, nil)
|
||||||
|
if err != nil {
|
||||||
|
log.Panicf("Failed to setup kafka producers(%s): %s", cfg.KafkaProducerURIs, err)
|
||||||
|
}
|
||||||
|
|
||||||
|
routing.Setup(http.DefaultServeMux, http.DefaultClient, cfg, producer)
|
||||||
log.Fatal(http.ListenAndServe(bindAddr, nil))
|
log.Fatal(http.ListenAndServe(bindAddr, nil))
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -4,7 +4,13 @@ import "golang.org/x/crypto/ed25519"
|
||||||
|
|
||||||
// ClientAPI contains the config information necessary to spin up a clientapi process.
|
// ClientAPI contains the config information necessary to spin up a clientapi process.
|
||||||
type ClientAPI struct {
|
type ClientAPI struct {
|
||||||
|
// The name of the server. This is usually the domain name, e.g 'matrix.org', 'localhost'.
|
||||||
ServerName string
|
ServerName string
|
||||||
|
// The private key which will be used to sign events.
|
||||||
PrivateKey ed25519.PrivateKey
|
PrivateKey ed25519.PrivateKey
|
||||||
KeyID string
|
// An arbitrary string used to uniquely identify the PrivateKey. Must start with the
|
||||||
|
// prefix "ed25519:".
|
||||||
|
KeyID string
|
||||||
|
// A list of URIs to send events to. These kafka logs should be consumed by a Room Server.
|
||||||
|
KafkaProducerURIs []string
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -9,13 +9,14 @@ import (
|
||||||
"github.com/matrix-org/dendrite/clientapi/writers"
|
"github.com/matrix-org/dendrite/clientapi/writers"
|
||||||
"github.com/matrix-org/util"
|
"github.com/matrix-org/util"
|
||||||
"github.com/prometheus/client_golang/prometheus"
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
|
sarama "gopkg.in/Shopify/sarama.v1"
|
||||||
)
|
)
|
||||||
|
|
||||||
const pathPrefixR0 = "/_matrix/client/r0"
|
const pathPrefixR0 = "/_matrix/client/r0"
|
||||||
|
|
||||||
// Setup registers HTTP handlers with the given ServeMux. It also supplies the given http.Client
|
// Setup registers HTTP handlers with the given ServeMux. It also supplies the given http.Client
|
||||||
// to clients which need to make outbound HTTP requests.
|
// to clients which need to make outbound HTTP requests.
|
||||||
func Setup(servMux *http.ServeMux, httpClient *http.Client, cfg config.ClientAPI) {
|
func Setup(servMux *http.ServeMux, httpClient *http.Client, cfg config.ClientAPI, producer sarama.SyncProducer) {
|
||||||
apiMux := mux.NewRouter()
|
apiMux := mux.NewRouter()
|
||||||
r0mux := apiMux.PathPrefix(pathPrefixR0).Subrouter()
|
r0mux := apiMux.PathPrefix(pathPrefixR0).Subrouter()
|
||||||
r0mux.Handle("/createRoom", make("createRoom", util.NewJSONRequestHandler(func(req *http.Request) util.JSONResponse {
|
r0mux.Handle("/createRoom", make("createRoom", util.NewJSONRequestHandler(func(req *http.Request) util.JSONResponse {
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue