forked from DiceDB/dice
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
DiceDB#553: Add Support for QWATCH Command with HTTP (DiceDB#614)
- Loading branch information
1 parent
79a97ee
commit 55160a2
Showing
16 changed files
with
572 additions
and
157 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,140 @@ | ||
| package http | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "encoding/json" | ||
| "github.com/stretchr/testify/assert" | ||
| "io" | ||
| "net/http" | ||
| "sync" | ||
| "testing" | ||
| "time" | ||
| ) | ||
|
|
||
| var qWatchQuery = "SELECT $key, $value WHERE $key LIKE \"match:100:*\" AND $value > 10 ORDER BY $value DESC LIMIT 3" | ||
|
|
||
| func TestQWatch(t *testing.T) { | ||
| exec := NewHTTPCommandExecutor() | ||
|
|
||
| testCases := []TestCase{ | ||
| { | ||
| name: "QWATCH Register Bad Request", | ||
| commands: []HTTPCommand{ | ||
| {Command: "QWATCH", Body: map[string]interface{}{}}, | ||
| }, | ||
| expected: []interface{}{ | ||
| []interface{}{}, | ||
| }, | ||
| errorExpected: true, | ||
| }, | ||
| { | ||
| name: "QWATCH Register", | ||
| commands: []HTTPCommand{ | ||
| {Command: "QWATCH", Body: map[string]interface{}{"query": qWatchQuery}}, | ||
| }, | ||
| expected: []interface{}{ | ||
| []interface{}{ | ||
| "qwatch", | ||
| "SELECT $key, $value WHERE $key like 'match:100:*' and $value > 10 ORDER BY $value desc LIMIT 3", | ||
| // Empty array, as the initial result will be empty | ||
| []interface{}{}, | ||
| }, | ||
| }, | ||
| errorExpected: false, | ||
| }, | ||
| } | ||
|
|
||
| for _, tc := range testCases { | ||
| t.Run(tc.name, func(t *testing.T) { | ||
| exec.FireCommand(HTTPCommand{ | ||
| Command: "DEL", | ||
| Body: map[string]interface{}{"key": "match:100:user"}, | ||
| }) | ||
|
|
||
| for i, cmd := range tc.commands { | ||
| result, err := exec.FireCommand(cmd) | ||
| if tc.errorExpected { | ||
| assert.NotNil(t, err) | ||
| } else { | ||
| assert.Equal(t, tc.expected[i], result) | ||
| } | ||
| } | ||
| }) | ||
| } | ||
| } | ||
|
|
||
| func TestQwatchWithSSE(t *testing.T) { | ||
| exec := NewHTTPCommandExecutor() | ||
| const key = "match:100:user:3" | ||
| const val = 15 | ||
|
|
||
| qwatchResponseReceived := make(chan struct{}, 2) | ||
| var wg sync.WaitGroup | ||
| wg.Add(1) | ||
|
|
||
| go func() { | ||
| defer wg.Done() | ||
|
|
||
| resp, err := http.Post("http://localhost:8083/qwatch", "application/json", | ||
| bytes.NewBuffer([]byte(`{ | ||
| "query": "SELECT $key, $value WHERE $key like 'match:100:*' and $value > 10 ORDER BY $value desc LIMIT 3" | ||
| }`))) | ||
| if err != nil { | ||
| t.Errorf("Failed to start QWATCH: %v", err) | ||
| return | ||
| } | ||
| defer func(Body io.ReadCloser) { | ||
| err := Body.Close() | ||
| if err != nil { | ||
| t.Errorf("Failed to close Resp body: %v", err) | ||
| } | ||
| }(resp.Body) | ||
|
|
||
| decoder := json.NewDecoder(resp.Body) | ||
| expectedResponses := []interface{}{ | ||
| []interface{}{ | ||
| "qwatch", | ||
| "SELECT $key, $value WHERE $key like 'match:100:*' and $value > 10 ORDER BY $value desc LIMIT 3", | ||
| []interface{}{}, | ||
| }, | ||
| map[string]interface{}{ | ||
| "cmd": "qwatch", | ||
| "query": "SELECT $key, $value WHERE $key like 'match:100:*' and $value > 10 ORDER BY $value desc LIMIT 3", | ||
| "data": []interface{}{[]interface{}{key, float64(val)}}, | ||
| }, | ||
| } | ||
|
|
||
| for responseCount := 0; responseCount < 2; responseCount++ { | ||
| var result interface{} | ||
| if err := decoder.Decode(&result); err != nil { | ||
| if err == io.EOF { | ||
| break | ||
| } | ||
| t.Errorf("Error reading SSE response: %v", err) | ||
| return | ||
| } | ||
|
|
||
| assert.Equal(t, result, expectedResponses[responseCount]) | ||
| qwatchResponseReceived <- struct{}{} | ||
| } | ||
| }() | ||
|
|
||
| time.Sleep(2 * time.Second) | ||
|
|
||
| setCmd := HTTPCommand{ | ||
| Command: "SET", | ||
| Body: map[string]interface{}{"key": key, "value": val}, | ||
| } | ||
| result, _ := exec.FireCommand(setCmd) | ||
| assert.Equal(t, result, "OK") | ||
|
|
||
| for i := 0; i < 2; i++ { | ||
| select { | ||
| case <-qwatchResponseReceived: | ||
| case <-time.After(10 * time.Second): | ||
| t.Fatalf("Timed out waiting for SSE response") | ||
| } | ||
| } | ||
|
|
||
| wg.Wait() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.