mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-10 16:33:11 -06:00
Code Issues Packages Projects Releases Wiki Activity

Get latest changes from master

Browse source
This commit is contained in:
Brendan Abolivier 2017-08-18 16:14:25 +01:00
parent f03bb7cf39 efbc14f7b9
commit 86c2174faf
2 changed files with 9 additions and 27 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
src/github.com/matrix-org/dendrite/clientapi/routing/routing.go
View file

@ -17,6 +17,7 @@ package routing
import (
"encoding/json"
"net/http"
"strings"
"github.com/gorilla/mux"
"github.com/matrix-org/dendrite/clientapi/auth/authtypes"
@ -93,18 +94,23 @@ func Setup(
return writers.SendEvent(req, device, vars["roomID"], vars["eventType"], vars["txnID"], nil, cfg, queryAPI, producer)
}),
)
r0mux.Handle("/rooms/{roomID}/state/{eventType}",
r0mux.Handle("/rooms/{roomID}/state/{eventType:[^/]+/?}",
common.MakeAuthAPI("send_message", deviceDB, func(req *http.Request, device *authtypes.Device) util.JSONResponse {
vars := mux.Vars(req)
emptyString := ""
return writers.SendEvent(req, device, vars["roomID"], vars["eventType"], vars["txnID"], &emptyString, cfg, queryAPI, producer)
eventType := vars["eventType"]
// If there's a trailing slash, remove it
if strings.HasSuffix(eventType, "/") {
eventType = eventType[:len(eventType)-1]
}
return writers.SendEvent(req, device, vars["roomID"], eventType, "", &emptyString, cfg, queryAPI, producer)
}),
)
r0mux.Handle("/rooms/{roomID}/state/{eventType}/{stateKey}",
common.MakeAuthAPI("send_message", deviceDB, func(req *http.Request, device *authtypes.Device) util.JSONResponse {
vars := mux.Vars(req)
stateKey := vars["stateKey"]
return writers.SendEvent(req, device, vars["roomID"], vars["eventType"], vars["txnID"], &stateKey, cfg, queryAPI, producer)
return writers.SendEvent(req, device, vars["roomID"], vars["eventType"], "", &stateKey, cfg, queryAPI, producer)
}),
)

24
src/github.com/matrix-org/dendrite/roomserver/input/input.go
View file

@ -17,9 +17,6 @@ package input
import (
"encoding/json"
"fmt"
"sync/atomic"
"net/http"
"github.com/matrix-org/dendrite/common"
@ -35,15 +32,6 @@ type RoomserverInputAPI struct {
// The kafkaesque topic to output new room events to.
// This is the name used in kafka to identify the stream to write events to.
OutputRoomEventTopic string
// If non-nil then the API will stop processing messages after this
// many messages and will shutdown. Malformed messages are not in the count.
StopProcessingAfter *int64
// If not-nil then the API will call this to shutdown the server.
// If this is nil then the API will continue to process messsages even
// though StopProcessingAfter has been reached.
ShutdownCallback func(reason string)
// How many messages the consumer has processed.
processed int64
}
// WriteOutputEvents implements OutputRoomEventWriter
@ -72,18 +60,6 @@ func (r *RoomserverInputAPI) InputRoomEvents(
if err := processRoomEvent(r.DB, r, request.InputRoomEvents[i]); err != nil {
return err
}
// Update the number of processed messages using atomic addition because it is accessed from multiple goroutines.
processed := atomic.AddInt64(&r.processed, 1)
// Check if we should stop processing.
// Note that since we have multiple goroutines it's quite likely that we'll overshoot by a few messages.
// If we try to stop processing after M message and we have N goroutines then we will process somewhere
// between M and (N + M) messages because the N goroutines could all try to process what they think will be the
// last message. We could be more careful here but this is good enough for getting rough benchmarks.
if r.StopProcessingAfter != nil && processed >= int64(*r.StopProcessingAfter) {
if r.ShutdownCallback != nil {
r.ShutdownCallback(fmt.Sprintf("Stopping processing after %d messages", r.processed))
}
}
}
return nil
}