Merge branch 'sse'

k6/loadtest_poll.k6.js

@@ -0,0 +1,48 @@
+import { check, fail, sleep } from "k6";
+import http from "k6/http";
+
+const publicKey = open("public.pem");
+
+export const options = {
+ scenarios: {
+ contacts: {
+ executor: "constant-vus",
+ vus: 10,
+ duration: "10s",
+ gracefulStop: "2m",
+ },
+ },
+};
+export default function () {
+ let formData = {
+ PUBLIC_KEY: publicKey,
+ };
+
+ let res = http.post("https://keelink.cloud/init.php", formData);
+
+ if (
+ !check(res, {
+ "is http_status 200 & status is 'true'": (r) =>
+ r.status === 200 && r.json()["status"] === true,
+ })
+ ) {
+ fail("failed to initialize");
+ }
+
+ const [sid, token] = res.json()["message"].split("###");
+
+ for (let i = 0; i < 60; i++) {
+ res = http.get(
+ `https://keelink.cloud/getcredforsid.php?sid=${sid}&token=${token}`
+ );
+
+ if (
+ !check(res, {
+ "no error thrown": (r) => r.status === 200,
+ })
+ ) {
+ fail("failed to read credentials");
+ }
+ sleep(1);
+ }
+}

k6/loadtest_sse.k6.js

@@ -0,0 +1,52 @@
+import { check, fail } from "k6";
+import http from "k6/http";
+
+const publicKey = open("public.pem");
+
+export const options = {
+ scenarios: {
+ contacts: {
+ executor: "constant-vus",
+ vus: 10,
+ duration: "1s",
+ gracefulStop: "2m",
+ },
+ },
+};
+export default function () {
+ let formData = {
+ PUBLIC_KEY: publicKey,
+ };
+
+ let res = http.post("https://keelink.cloud/init.php", formData);
+
+ if (
+ !check(res, {
+ "is http_status 200 & status is 'true'": (r) =>
+ r.status === 200 && r.json()["status"] === true,
+ })
+ ) {
+ fail("failed to initialize");
+ }
+
+ const [sid, token] = res.json()["message"].split("###");
+
+ res = http.get(
+ `https://keelink.cloud/getcredforsid.php?sid=${sid}&token=${token}`,
+ {
+ headers: {
+ Accept: "text/event-stream",
+ },
+ timeout: "120s",
+ }
+ );
+
+ if (
+ !check(res, {
+ "no error thrown": (r) =>
+ r.error === "" && (r.status === 0 || r.status === 200),
+ })
+ ) {
+ fail("failed to read credentials");
+ }
+}

k6/public.pem

@@ -0,0 +1 @@
+MIIBITANBgkqhkiG9w0BAQEFAAOCAQ4AMIIBCQKCAQBqYWgKLPUHxJwJfZ5Z_VpOGrojfN23zFqVuQgAVmZRvYJQ2wUklB0sjerJzIdLezfjd4QJo5tXHCyitvcvJwapFM1T_OQDmcIDrULzEIwNVAgapTMRPEEIjNhrbOklBX-uERWEz6arOawo75Ydr-GexZR0Fx7kph-c9E490mZyNfneyANuSpnqcS1RkMyJ6ekRnUQcPkYM64hxIL_5woiqGQtoDECOtNAo6AWqV8-X1-7ZEWVBFd44YjKjxDZO62abrBqMESRS2QoQqM0PYPrAMUA_vMGLKBg6RkxuoIG19ayIyDTF3aG4A-3VOyvkNnmyeQH3NrdQ__OaxTnWcCfNAgMBAAE=

webapp/api/get_cred_for_sid.go

@@ -4,6 +4,7 @@ import (
  "io"
  "net/http"
  "strings"
+ "sync"
  "time"
 
  "github.com/gin-gonic/gin"
@@ -47,16 +48,33 @@ func handleLegacy(c *gin.Context) {
 
 func handleWithSSE(c *gin.Context) {
  clientChan := make(chan sseCredForSessionIDResponse)
+ wg := sync.WaitGroup{}
+ var clientDisconnected bool
+
+ defer func() {
+ glg.Debugf("closing channel")
+ close(clientChan)
+ }()
+
  c.Writer.Header().Set("Content-Type", "text/event-stream")
  c.Writer.Header().Set("Cache-Control", "no-cache")
  c.Writer.Header().Set("Connection", "keep-alive")
  c.Writer.Header().Set("Transfer-Encoding", "chunked")
  c.Writer.Flush()
 
+ wg.Add(1)
  go func() {
+ defer func() {
+ wg.Done()
+ }()
  timeout := time.Now().Add(time.Minute)
 
  for {
+
+ if clientDisconnected {
+ return
+ }
+
  entry, sid, found := getEntryFromSessionID(c, true)
 
  if time.Now().Compare(timeout) > 0 {
@@ -68,41 +86,38 @@ func handleWithSSE(c *gin.Context) {
  if !found {
  clientChan <- sseCredForSessionIDResponse{Status: false, Message: "Credentials not yet received"}
  time.Sleep(time.Second * 2)
- continue
- }
 
- if entry.EncryptedPassword == nil && entry.Username == nil {
+ } else if entry.EncryptedPassword == nil && entry.Username == nil {
  glg.Debugf("Credentials not yet received for session ID: %s", sid)
 
  clientChan <- sseCredForSessionIDResponse{Status: false, Message: "Credentials not yet received, empty"}
  time.Sleep(time.Second * 2)
  continue
+ } else {
+ clientChan <- sseCredForSessionIDResponse{Status: true, Username: entry.Username, EncryptedPassword: entry.EncryptedPassword}
+ return
  }
-
- clientChan <- sseCredForSessionIDResponse{Status: true, Username: entry.Username, EncryptedPassword: entry.EncryptedPassword}
- return
  }
 
  }()
 
- clientDisconnected := c.Stream(func(w io.Writer) bool {
+ clientDisconnected = c.Stream(func(w io.Writer) bool {
  // Stream message to client from message channel
  if response, ok := <-clientChan; ok {
  if response.Status {
  c.SSEvent("message", gin.H{"status": response.Status, "username": response.Username, "password": response.EncryptedPassword})
+ return false
  } else {
  c.SSEvent("message", gin.H{"status": response.Status, "message": response.Message})
+ return true
  }
-
- if response.TimeoutReached {
- close(clientChan)
- }
-
- return true
  }
  return false
  })
 
+ // Wait for goroutine to finish before closing the channel
+ wg.Wait()
+
  glg.Debugf("client disconnect: %t", clientDisconnected)
 }