start working at worker groups

broker/src/asapo_broker/database/database.go

 package database
 
 type Agent interface {
-	GetRecordFromDb(db_name string, op string, id int) ([]byte, error)
+	GetRecordFromDb(db_name string, group_id string, op string, id int) ([]byte, error)
 	Connect(string) error
 	Close()
 	Copy() Agent

broker/src/asapo_broker/database/mock_database.go

@@ -24,7 +24,7 @@ func (db *MockedDatabase) Copy() Agent {
 	return db
 }
 
-func (db *MockedDatabase) GetRecordFromDb(db_name string, op string, id int) (answer []byte, err error) {
-	args := db.Called(db_name, op, id)
+func (db *MockedDatabase) GetRecordFromDb(db_name string, group_id string, op string, id int) (answer []byte, err error) {
+	args := db.Called(db_name, group_id, op, id)
 	return args.Get(0).([]byte), args.Error(1)
 }

broker/src/asapo_broker/database/mongodb.go

@@ -127,32 +127,32 @@ func (db *Mongodb) getMaxIndex(dbname string) (max_id int, err error) {
 	return id.ID, nil
 }
 
-func (db *Mongodb) createLocationPointers(dbname string) (err error) {
+func (db *Mongodb) createLocationPointers(dbname string, group_id string) (err error) {
 	change := mgo.Change{
 		Update: bson.M{"$inc": bson.M{pointer_field_name: 0}},
 		Upsert: true,
 	}
-	q := bson.M{"_id": 0}
+	q := bson.M{"_id": group_id}
 	c := db.session.DB(dbname).C(pointer_collection_name)
 	var res map[string]interface{}
 	_, err = c.Find(q).Apply(change, &res)
 	return err
 }
 
-func (db *Mongodb) setCounter(dbname string, ind int) (err error) {
+func (db *Mongodb) setCounter(dbname string, group_id string, ind int) (err error) {
 	update := bson.M{"$set": bson.M{pointer_field_name: ind}}
 	c := db.session.DB(dbname).C(pointer_collection_name)
-	return c.UpdateId(0, update)
+	return c.UpdateId(group_id, update)
 }
 
-func (db *Mongodb) incrementField(dbname string, max_ind int, res interface{}) (err error) {
+func (db *Mongodb) incrementField(dbname string, group_id string, max_ind int, res interface{}) (err error) {
 	update := bson.M{"$inc": bson.M{pointer_field_name: 1}}
 	change := mgo.Change{
 		Update:    update,
 		Upsert:    false,
 		ReturnNew: true,
 	}
-	q := bson.M{"_id": 0, pointer_field_name: bson.M{"$lt": max_ind}}
+	q := bson.M{"_id": group_id, pointer_field_name: bson.M{"$lt": max_ind}}
 	c := db.session.DB(dbname).C(pointer_collection_name)
 	_, err = c.Find(q).Apply(change, res)
 	if err == mgo.ErrNotFound {
@@ -186,26 +186,26 @@ func (db *Mongodb) GetRecordByID(dbname string, id int, returnID bool) ([]byte,
 	return utils.MapToJson(&res)
 }
 
-func (db *Mongodb) needCreateLocationPointersInDb(db_name string) bool {
+func (db *Mongodb) needCreateLocationPointersInDb(group_id string) bool {
 	dbPointersLock.RLock()
-	needCreate := !db.db_pointers_created[db_name]
+	needCreate := !db.db_pointers_created[group_id]
 	dbPointersLock.RUnlock()
 	return needCreate
 }
 
-func (db *Mongodb) SetLocationPointersCreateFlag(db_name string) {
+func (db *Mongodb) SetLocationPointersCreateFlag(group_id string) {
 	dbPointersLock.Lock()
 	if db.db_pointers_created == nil {
 		db.db_pointers_created = make(map[string]bool)
 	}
-	db.db_pointers_created[db_name] = true
+	db.db_pointers_created[group_id] = true
 	dbPointersLock.Unlock()
 }
 
-func (db *Mongodb) generateLocationPointersInDbIfNeeded(db_name string) {
-	if db.needCreateLocationPointersInDb(db_name) {
-		db.createLocationPointers(db_name)
-		db.SetLocationPointersCreateFlag(db_name)
+func (db *Mongodb) generateLocationPointersInDbIfNeeded(db_name string, group_id string) {
+	if db.needCreateLocationPointersInDb(group_id) {
+		db.createLocationPointers(db_name, group_id)
+		db.SetLocationPointersCreateFlag(group_id)
 	}
 }
 
@@ -217,7 +217,7 @@ func (db *Mongodb) getParentDB() *Mongodb {
 	}
 }
 
-func (db *Mongodb) checkDatabaseOperationPrerequisites(db_name string) error {
+func (db *Mongodb) checkDatabaseOperationPrerequisites(db_name string, group_id string) error {
 	if db.session == nil {
 		return &DBError{utils.StatusError, no_session_msg}
 	}
@@ -226,18 +226,18 @@ func (db *Mongodb) checkDatabaseOperationPrerequisites(db_name string) error {
 		return &DBError{utils.StatusWrongInput, err.Error()}
 	}
 
-	db.getParentDB().generateLocationPointersInDbIfNeeded(db_name)
+	db.getParentDB().generateLocationPointersInDbIfNeeded(db_name, group_id)
 
 	return nil
 }
 
-func (db *Mongodb) getCurrentPointer(db_name string) (Pointer, error) {
+func (db *Mongodb) getCurrentPointer(db_name string, group_id string) (Pointer, error) {
 	max_ind, err := db.getMaxIndex(db_name)
 	if err != nil {
 		return Pointer{}, err
 	}
 	var curPointer Pointer
-	err = db.incrementField(db_name, max_ind, &curPointer)
+	err = db.incrementField(db_name, group_id, max_ind, &curPointer)
 	if err != nil {
 		return Pointer{}, err
 	}
@@ -245,51 +245,51 @@ func (db *Mongodb) getCurrentPointer(db_name string) (Pointer, error) {
 	return curPointer, nil
 }
 
-func (db *Mongodb) GetNextRecord(db_name string) ([]byte, error) {
+func (db *Mongodb) GetNextRecord(db_name string, group_id string) ([]byte, error) {
 
-	if err := db.checkDatabaseOperationPrerequisites(db_name); err != nil {
+	if err := db.checkDatabaseOperationPrerequisites(db_name, group_id); err != nil {
 		return nil, err
 	}
 
-	curPointer, err := db.getCurrentPointer(db_name)
+	curPointer, err := db.getCurrentPointer(db_name, group_id)
 	if err != nil {
-		log_str := "error getting next pointer for " + db_name + ":" + err.Error()
+		log_str := "error getting next pointer for " + db_name + ", groupid: " + group_id + ":" + err.Error()
 		logger.Debug(log_str)
 		return nil, err
 	}
-	log_str := "got next pointer " + strconv.Itoa(curPointer.Value) + " for " + db_name
+	log_str := "got next pointer " + strconv.Itoa(curPointer.Value) + " for " + db_name + ", groupid: " + group_id
 	logger.Debug(log_str)
 	return db.GetRecordByID(db_name, curPointer.Value, true)
 
 }
 
-func (db *Mongodb) GetLastRecord(db_name string) ([]byte, error) {
+func (db *Mongodb) GetLastRecord(db_name string, group_id string) ([]byte, error) {
 
-	if err := db.checkDatabaseOperationPrerequisites(db_name); err != nil {
+	if err := db.checkDatabaseOperationPrerequisites(db_name, group_id); err != nil {
 		return nil, err
 	}
 
 	max_ind, err := db.getMaxIndex(db_name)
 	if err != nil {
-		log_str := "error getting last pointer for " + db_name + ":" + err.Error()
+		log_str := "error getting last pointer for " + db_name + ", groupid: " + group_id + ":" + err.Error()
 		logger.Debug(log_str)
 		return nil, err
 	}
 	res, err := db.GetRecordByID(db_name, max_ind, false)
 
-	db.setCounter(db_name, max_ind)
+	db.setCounter(db_name, group_id, max_ind)
 
 	return res, err
 }
 
-func (db *Mongodb) GetRecordFromDb(db_name string, op string, id int) (answer []byte, err error) {
+func (db *Mongodb) GetRecordFromDb(db_name string, group_id string, op string, id int) (answer []byte, err error) {
 	switch op {
 	case "next":
-		return db.GetNextRecord(db_name)
+		return db.GetNextRecord(db_name, group_id)
 	case "id":
 		return db.GetRecordByID(db_name, id, true)
 	case "last":
-		return db.GetLastRecord(db_name)
+		return db.GetLastRecord(db_name, group_id)
 	}
 	return nil, errors.New("Wrong db operation: " + op)
 }

broker/src/asapo_broker/database/mongodb_test.go

@@ -19,6 +19,7 @@ var db Mongodb
 
 const dbname = "run1"
 const dbaddress = "127.0.0.1:27017"
+const groupId = "bid2a5auidddp1vl71d0"
 
 var rec1 = TestRecord{1, "aaa"}
 var rec2 = TestRecord{2, "bbb"}
@@ -49,14 +50,14 @@ func TestMongoDBConnectOK(t *testing.T) {
 }
 
 func TestMongoDBGetNextErrorWhenNotConnected(t *testing.T) {
-	_, err := db.GetNextRecord("")
+	_, err := db.GetNextRecord("", groupId)
 	assert.Equal(t, utils.StatusError, err.(*DBError).Code)
 }
 
 func TestMongoDBGetNextErrorWhenWrongDatabasename(t *testing.T) {
 	db.Connect(dbaddress)
 	defer cleanup()
-	_, err := db.GetNextRecord("")
+	_, err := db.GetNextRecord("", groupId)
 	assert.Equal(t, utils.StatusWrongInput, err.(*DBError).Code)
 }
 
@@ -64,7 +65,7 @@ func TestMongoDBGetNextErrorWhenEmptyCollection(t *testing.T) {
 	db.Connect(dbaddress)
 	db.databases = append(db.databases, dbname)
 	defer cleanup()
-	_, err := db.GetNextRecord(dbname)
+	_, err := db.GetNextRecord(dbname, groupId)
 	assert.Equal(t, utils.StatusNoData, err.(*DBError).Code)
 }
 
@@ -72,7 +73,7 @@ func TestMongoDBGetNextErrorWhenRecordNotThereYet(t *testing.T) {
 	db.Connect(dbaddress)
 	defer cleanup()
 	db.InsertRecord(dbname, &rec2)
-	_, err := db.GetNextRecord(dbname)
+	_, err := db.GetNextRecord(dbname, groupId)
 	assert.Equal(t, utils.StatusNoData, err.(*DBError).Code)
 	assert.Equal(t, "{\"id\":1}", err.Error())
 }
@@ -81,7 +82,7 @@ func TestMongoDBGetNextOK(t *testing.T) {
 	db.Connect(dbaddress)
 	defer cleanup()
 	db.InsertRecord(dbname, &rec1)
-	res, err := db.GetNextRecord(dbname)
+	res, err := db.GetNextRecord(dbname, groupId)
 	assert.Nil(t, err)
 	assert.Equal(t, string(rec1_expect), string(res))
 }
@@ -90,8 +91,8 @@ func TestMongoDBGetNextErrorOnNoMoreData(t *testing.T) {
 	db.Connect(dbaddress)
 	defer cleanup()
 	db.InsertRecord(dbname, &rec1)
-	db.GetNextRecord(dbname)
-	_, err := db.GetNextRecord(dbname)
+	db.GetNextRecord(dbname, groupId)
+	_, err := db.GetNextRecord(dbname, groupId)
 	assert.Equal(t, utils.StatusNoData, err.(*DBError).Code)
 }
 
@@ -100,8 +101,8 @@ func TestMongoDBGetNextCorrectOrder(t *testing.T) {
 	defer cleanup()
 	db.InsertRecord(dbname, &rec2)
 	db.InsertRecord(dbname, &rec1)
-	res1, _ := db.GetNextRecord(dbname)
-	res2, _ := db.GetNextRecord(dbname)
+	res1, _ := db.GetNextRecord(dbname, groupId)
+	res2, _ := db.GetNextRecord(dbname, groupId)
 	assert.Equal(t, string(rec1_expect), string(res1))
 	assert.Equal(t, string(rec2_expect), string(res2))
 }
@@ -133,7 +134,7 @@ func getRecords(n int) []int {
 	for i := 0; i < n; i++ {
 		go func() {
 			defer wg.Done()
-			res_bin, _ := db.GetNextRecord(dbname)
+			res_bin, _ := db.GetNextRecord(dbname, groupId)
 			var res TestRecord
 			json.Unmarshal(res_bin, &res)
 			results[res.ID] = 1
@@ -177,7 +178,7 @@ func TestMongoDBGetRecordNext(t *testing.T) {
 	db.Connect(dbaddress)
 	defer cleanup()
 	db.InsertRecord(dbname, &rec1)
-	res, err := db.GetRecordFromDb(dbname, "next", 0)
+	res, err := db.GetRecordFromDb(dbname, groupId, "next", 0)
 	assert.Nil(t, err)
 	assert.Equal(t, string(rec1_expect), string(res))
 }
@@ -186,7 +187,7 @@ func TestMongoDBGetRecordID(t *testing.T) {
 	db.Connect(dbaddress)
 	defer cleanup()
 	db.InsertRecord(dbname, &rec1)
-	res, err := db.GetRecordFromDb(dbname, "id", 1)
+	res, err := db.GetRecordFromDb(dbname, groupId, "id", 1)
 	assert.Nil(t, err)
 	assert.Equal(t, string(rec1_expect), string(res))
 }
@@ -195,7 +196,7 @@ func TestMongoDBWrongOp(t *testing.T) {
 	db.Connect(dbaddress)
 	defer cleanup()
 	db.InsertRecord(dbname, &rec1)
-	_, err := db.GetRecordFromDb(dbname, "bla", 0)
+	_, err := db.GetRecordFromDb(dbname, groupId, "bla", 0)
 	assert.NotNil(t, err)
 }
 
@@ -205,7 +206,7 @@ func TestMongoDBGetRecordLast(t *testing.T) {
 	db.InsertRecord(dbname, &rec1)
 	db.InsertRecord(dbname, &rec2)
 
-	res, err := db.GetRecordFromDb(dbname, "last", 0)
+	res, err := db.GetRecordFromDb(dbname, groupId, "last", 0)
 	assert.Nil(t, err)
 	assert.Equal(t, string(rec2_expect), string(res))
 }
@@ -216,13 +217,13 @@ func TestMongoDBGetNextAfterGetLastCorrect(t *testing.T) {
 	db.InsertRecord(dbname, &rec1)
 	db.InsertRecord(dbname, &rec2)
 
-	res, err := db.GetRecordFromDb(dbname, "last", 0)
+	res, err := db.GetRecordFromDb(dbname, groupId, "last", 0)
 	assert.Nil(t, err)
 	assert.Equal(t, string(rec2_expect), string(res))
 
 	db.InsertRecord(dbname, &rec3)
 
-	res, err = db.GetRecordFromDb(dbname, "next", 0)
+	res, err = db.GetRecordFromDb(dbname, groupId, "next", 0)
 	assert.Nil(t, err)
 	assert.Equal(t, string(rec3_expect), string(res))
 

broker/src/asapo_broker/server/get_id.go

@@ -22,5 +22,5 @@ func routeGetByID(w http.ResponseWriter, r *http.Request) {
 		w.WriteHeader(http.StatusBadRequest)
 		return
 	}
-	getImage(w, r, "id", id)
+	getImage(w, r, "id", id, false)
 }

broker/src/asapo_broker/server/get_id_test.go

@@ -45,7 +45,7 @@ func TestGetIDTestSuite(t *testing.T) {
 }
 
 func (suite *GetIDTestSuite) TestGetIdCallsCorrectRoutine() {
-	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, "id", 1).Return([]byte("Hello"), nil)
+	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, "", "id", 1).Return([]byte("Hello"), nil)
 	logger.MockLog.On("Debug", mock.MatchedBy(containsMatcher("get id request")))
 	ExpectCopyClose(suite.mock_db)
 

broker/src/asapo_broker/server/get_image.go

@@ -4,19 +4,26 @@ import (
 	"asapo_broker/database"
 	"asapo_common/logger"
 	"asapo_common/utils"
+	"fmt"
 	"github.com/gorilla/mux"
+	"github.com/rs/xid"
 	"net/http"
 )
 
-func extractRequestParameters(r *http.Request) (string, bool) {
+func extractRequestParameters(r *http.Request, needGroupID bool) (string, string, bool) {
 	vars := mux.Vars(r)
-	db_name, ok := vars["dbname"]
-	return db_name, ok
+	db_name, ok1 := vars["dbname"]
+	ok2 := true
+	group_id := ""
+	if needGroupID {
+		group_id, ok2 = vars["groupid"]
+	}
+	return db_name, group_id, ok1 && ok2
 }
 
-func getImage(w http.ResponseWriter, r *http.Request, op string, id int) {
+func getImage(w http.ResponseWriter, r *http.Request, op string, id int, needGroupID bool) {
 	r.Header.Set("Content-type", "application/json")
-	db_name, ok := extractRequestParameters(r)
+	db_name, group_id, ok := extractRequestParameters(r, needGroupID)
 	if !ok {
 		w.WriteHeader(http.StatusBadRequest)
 		return
@@ -27,7 +34,18 @@ func getImage(w http.ResponseWriter, r *http.Request, op string, id int) {
 		return
 	}
 
-	answer, code := getRecord(db_name, op, id)
+	if needGroupID {
+		if _, err := xid.FromString(group_id); err != nil {
+			err_str := "wrong groupid " + group_id
+			log_str := "processing get " + op + " request in " + db_name + " at " + settings.BrokerDbAddress + ": " + err_str
+			logger.Error(log_str)
+			fmt.Println(log_str)
+			w.WriteHeader(http.StatusBadRequest)
+			w.Write([]byte(err_str))
+			return
+		}
+	}
+	answer, code := getRecord(db_name, group_id, op, id)
 	w.WriteHeader(code)
 	w.Write(answer)
 }
@@ -46,11 +64,11 @@ func returnError(err error, log_str string) (answer []byte, code int) {
 	return []byte(err.Error()), code
 }
 
-func getRecord(db_name string, op string, id int) (answer []byte, code int) {
+func getRecord(db_name string, group_id string, op string, id int) (answer []byte, code int) {
 	db_new := db.Copy()
 	defer db_new.Close()
 	statistics.IncreaseCounter()
-	answer, err := db_new.GetRecordFromDb(db_name, op, id)
+	answer, err := db_new.GetRecordFromDb(db_name, group_id, op, id)
 	log_str := "processing get " + op + " request in " + db_name + " at " + settings.BrokerDbAddress
 	if err != nil {
 		return returnError(err, log_str)

broker/src/asapo_broker/server/get_image_test.go

@@ -16,6 +16,9 @@ import (
 
 var correctTokenSuffix, wrongTokenSuffix, suffixWithWrongToken, expectedBeamtimeId string
 
+const expectedGroupID = "bid2a5auidddp1vl71d0"
+const wrongGroupID = "bid2a5auidddp1vl71"
+
 func prepareTestAuth() {
 	expectedBeamtimeId = "beamtime_id"
 	auth = utils.NewHMACAuth("secret")
@@ -46,9 +49,14 @@ func containsMatcher(substrings ...string) func(str string) bool {
 	}
 }
 
-func doRequest(path string) *httptest.ResponseRecorder {
+func doRequest(path string, method ...string) *httptest.ResponseRecorder {
+	m := "GET"
+	if len(method) > 0 {
+		m = method[0]
+	}
+
 	mux := utils.NewRouter(listRoutes)
-	req, _ := http.NewRequest("GET", path, nil)
+	req, _ := http.NewRequest(m, path, nil)
 	w := httptest.NewRecorder()
 	mux.ServeHTTP(w, req)
 	return w
@@ -90,7 +98,7 @@ func TestGetImageTestSuite(t *testing.T) {
 func (suite *GetImageTestSuite) TestGetImageWithWrongToken() {
 	logger.MockLog.On("Error", mock.MatchedBy(containsMatcher("wrong token")))
 
-	w := doRequest("/database/" + expectedBeamtimeId + "/next" + suffixWithWrongToken)
+	w := doRequest("/database/" + expectedBeamtimeId + "/" + expectedGroupID + "/next" + suffixWithWrongToken)
 
 	suite.Equal(http.StatusUnauthorized, w.Code, "wrong token")
 }
@@ -98,37 +106,43 @@ func (suite *GetImageTestSuite) TestGetImageWithWrongToken() {
 func (suite *GetImageTestSuite) TestGetImageWithNoToken() {
 	logger.MockLog.On("Error", mock.MatchedBy(containsMatcher("cannot extract")))
 
-	w := doRequest("/database/" + expectedBeamtimeId + "/next" + wrongTokenSuffix)
+	w := doRequest("/database/" + expectedBeamtimeId + "/" + expectedGroupID + "/next" + wrongTokenSuffix)
 
 	suite.Equal(http.StatusUnauthorized, w.Code, "no token")
 }
 
 func (suite *GetImageTestSuite) TestGetImageWithWrongDatabaseName() {
-	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, "next", 0).Return([]byte(""),
+	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, expectedGroupID, "next", 0).Return([]byte(""),
 		&database.DBError{utils.StatusWrongInput, ""})
 
 	logger.MockLog.On("Error", mock.MatchedBy(containsMatcher("get next request")))
 	ExpectCopyClose(suite.mock_db)
 
-	w := doRequest("/database/" + expectedBeamtimeId + "/next" + correctTokenSuffix)
+	w := doRequest("/database/" + expectedBeamtimeId + "/" + expectedGroupID + "/next" + correctTokenSuffix)
 
 	suite.Equal(http.StatusBadRequest, w.Code, "wrong database name")
 }
 
 func (suite *GetImageTestSuite) TestGetImageWithInternalDBError() {
-	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, "next", 0).Return([]byte(""), errors.New(""))
+	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, expectedGroupID, "next", 0).Return([]byte(""), errors.New(""))
 	logger.MockLog.On("Error", mock.MatchedBy(containsMatcher("get next request")))
 	ExpectCopyClose(suite.mock_db)
 
-	w := doRequest("/database/" + expectedBeamtimeId + "/next" + correctTokenSuffix)
+	w := doRequest("/database/" + expectedBeamtimeId + "/" + expectedGroupID + "/next" + correctTokenSuffix)
 	suite.Equal(http.StatusInternalServerError, w.Code, "internal error")
 }
 
 func (suite *GetImageTestSuite) TestGetImageAddsCounter() {
-	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, "next", 0).Return([]byte("Hello"), nil)
+	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, expectedGroupID, "next", 0).Return([]byte("Hello"), nil)
 	logger.MockLog.On("Debug", mock.MatchedBy(containsMatcher("get next request in "+expectedBeamtimeId)))
 	ExpectCopyClose(suite.mock_db)
 
-	doRequest("/database/" + expectedBeamtimeId + "/next" + correctTokenSuffix)
+	doRequest("/database/" + expectedBeamtimeId + "/" + expectedGroupID + "/next" + correctTokenSuffix)
 	suite.Equal(1, statistics.GetCounter(), "GetImage increases counter")
 }
+
+func (suite *GetImageTestSuite) TestGetImageWrongGroupID() {
+	logger.MockLog.On("Error", mock.MatchedBy(containsMatcher("wrong groupid")))
+	w := doRequest("/database/" + expectedBeamtimeId + "/" + wrongGroupID + "/next" + correctTokenSuffix)
+	suite.Equal(http.StatusBadRequest, w.Code, "wrong group id")
+}

broker/src/asapo_broker/server/get_last.go

@@ -5,5 +5,5 @@ import (
 )
 
 func routeGetLast(w http.ResponseWriter, r *http.Request) {
-	getImage(w, r, "last", 0)
+	getImage(w, r, "last", 0, true)
 }

broker/src/asapo_broker/server/get_last_test.go

@@ -33,11 +33,11 @@ func TestGetLastTestSuite(t *testing.T) {
 }
 
 func (suite *GetLastTestSuite) TestGetLastCallsCorrectRoutine() {
-	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, "last", 0).Return([]byte("Hello"), nil)
+	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, expectedGroupID, "last", 0).Return([]byte("Hello"), nil)
 	logger.MockLog.On("Debug", mock.MatchedBy(containsMatcher("get last request")))
 	ExpectCopyClose(suite.mock_db)
 
-	w := doRequest("/database/" + expectedBeamtimeId + "/last" + correctTokenSuffix)
+	w := doRequest("/database/" + expectedBeamtimeId + "/" + expectedGroupID + "/last" + correctTokenSuffix)
 	suite.Equal(http.StatusOK, w.Code, "GetLast OK")
 	suite.Equal("Hello", string(w.Body.Bytes()), "GetLast sends data")
 }

broker/src/asapo_broker/server/get_next.go

@@ -5,5 +5,5 @@ import (
 )
 
 func routeGetNext(w http.ResponseWriter, r *http.Request) {
-	getImage(w, r, "next", 0)
+	getImage(w, r, "next", 0, true)
 }

broker/src/asapo_broker/server/get_next_test.go

@@ -33,11 +33,11 @@ func TestGetNextTestSuite(t *testing.T) {
 }
 
 func (suite *GetNextTestSuite) TestGetNextCallsCorrectRoutine() {
-	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, "next", 0).Return([]byte("Hello"), nil)
+	suite.mock_db.On("GetRecordFromDb", expectedBeamtimeId, expectedGroupID, "next", 0).Return([]byte("Hello"), nil)
 	logger.MockLog.On("Debug", mock.MatchedBy(containsMatcher("get next request")))
 	ExpectCopyClose(suite.mock_db)
 
-	w := doRequest("/database/" + expectedBeamtimeId + "/next" + correctTokenSuffix)
+	w := doRequest("/database/" + expectedBeamtimeId + "/" + expectedGroupID + "/next" + correctTokenSuffix)
 	suite.Equal(http.StatusOK, w.Code, "GetNext OK")
 	suite.Equal("Hello", string(w.Body.Bytes()), "GetNext sends data")
 }

broker/src/asapo_broker/server/listroutes.go

@@ -8,13 +8,13 @@ var listRoutes = utils.Routes{
 	utils.Route{
 		"GetNext",
 		"Get",
-		"/database/{dbname}/next",
+		"/database/{dbname}/{groupid}/next",
 		routeGetNext,
 	},
 	utils.Route{
 		"GetLast",
 		"Get",
-		"/database/{dbname}/last",
+		"/database/{dbname}/{groupid}/last",
 		routeGetLast,
 	},
 	utils.Route{
@@ -23,6 +23,12 @@ var listRoutes = utils.Routes{
 		"/database/{dbname}/{id}",
 		routeGetByID,
 	},
+	utils.Route{
+		"GetID",
+		"Post",
+		"/creategroup",
+		routeCreateGroupID,
+	},
 
 	utils.Route{
 		"Health",

broker/src/asapo_broker/server/post_create_group.go

+package server
+
+import (
+	"asapo_common/logger"
+	"github.com/rs/xid"
+	"net/http"
+)
+
+func routeCreateGroupID(w http.ResponseWriter, r *http.Request) {
+	guid := xid.New()
+	w.Write([]byte(guid.String()))
+	logger.Debug("generated new group: " + guid.String())
+	statistics.IncreaseCounter()
+}

broker/src/asapo_broker/server/post_create_group_test.go

+package server
+
+import (
+	"asapo_common/logger"
+	"github.com/rs/xid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"net/http"
+	"testing"
+)
+
+func GetObjectID(t *testing.T) (xid.ID, error) {
+	w := doRequest("/creategroup", "POST")
+	assert.Equal(t, http.StatusOK, w.Code, "New Group OK")
+	return xid.FromString(w.Body.String())
+}
+
+func TestGetNewGroup(t *testing.T) {
+	statistics.Reset()
+	logger.SetMockLog()
+	logger.MockLog.On("Debug", mock.MatchedBy(containsMatcher("generated new group")))
+
+	id1, err := GetObjectID(t)
+	assert.Nil(t, err, "first is ObjectID")
+
+	id2, err := GetObjectID(t)
+	assert.Nil(t, err, "second is ObjectID")
+
+	assert.NotEqual(t, id1.String(), id2.String())
+	assert.Equal(t, id1.Counter()+1, id2.Counter())
+	assert.Equal(t, 2, statistics.GetCounter(), "creategroup increases counter")
+
+	logger.UnsetMockLog()
+}

tests/automatic/broker/check_monitoring/check_linux.sh

@@ -24,13 +24,16 @@ sleep 0.3
 
 brokerid=`echo $!`
 
+groupid=`curl -d '' --silent 127.0.0.1:5005/creategroup`
+
+
 for i in `seq 1 50`;
 do
-    curl --silent 127.0.0.1:5005/database/data/next?token=$token >/dev/null 2>&1 &
+    curl --silent 127.0.0.1:5005/database/data/${groupid}/next?token=$token >/dev/null 2>&1 &
 done
 
 
 sleep 3
 
-influx -execute "select sum(rate) from RequestsRate" -database=${database_name} -format=json | jq .results[0].series[0].values[0][1] | tee /dev/stderr | grep 50
+influx -execute "select sum(rate) from RequestsRate" -database=${database_name} -format=json | jq .results[0].series[0].values[0][1] | tee /dev/stderr | grep 51
 

tests/automatic/broker/get_last/check_linux.sh

@@ -23,16 +23,18 @@ sleep 0.3
 brokerid=`echo $!`
 
 
-curl -v  --silent 127.0.0.1:5005/database/data/last?token=$token --stderr -
+groupid=`curl -d '' --silent 127.0.0.1:5005/creategroup`
 
-curl -v  --silent 127.0.0.1:5005/database/data/last?token=$token --stderr - | grep '"_id":2'
-curl -v  --silent 127.0.0.1:5005/database/data/last?token=$token --stderr - | grep '"_id":2'
+curl -v  --silent 127.0.0.1:5005/database/data/${groupid}/last?token=$token --stderr -
+
+curl -v  --silent 127.0.0.1:5005/database/data/${groupid}/last?token=$token --stderr - | grep '"_id":2'
+curl -v  --silent 127.0.0.1:5005/database/data/${groupid}/last?token=$token --stderr - | grep '"_id":2'
 
 echo "db.data.insert({"_id":3})" | mongo ${database_name}
 
-curl -v  --silent 127.0.0.1:5005/database/data/last?token=$token --stderr - | grep '"_id":3'
+curl -v  --silent 127.0.0.1:5005/database/data/${groupid}/last?token=$token --stderr - | grep '"_id":3'
 
 echo "db.data.insert({"_id":4})" | mongo ${database_name}
 
-curl -v  --silent 127.0.0.1:5005/database/data/next?token=$token --stderr - | grep '"_id":4'
-curl -v  --silent 127.0.0.1:5005/database/data/last?token=$token --stderr - | grep '"_id":4'
\ No newline at end of file
+curl -v  --silent 127.0.0.1:5005/database/data/${groupid}/next?token=$token --stderr - | grep '"_id":4'
+curl -v  --silent 127.0.0.1:5005/database/data/${groupid}/last?token=$token --stderr - | grep '"_id":4'
\ No newline at end of file

tests/automatic/broker/get_last/check_windows.bat

@@ -10,22 +10,24 @@ set short_name="%~nx1"
 "%2" token -secret broker_secret.key data > token
 set /P token=< token
 
+curl -d '' --silent 127.0.0.1:5005/creategroup > groupid
+set /P groupid=< groupid
 
 
 start /B "" "%full_name%" -config settings.json
 
 ping 1.0.0.0 -n 1 -w 100 > nul
 
-C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/last?token=%token% --stderr - | findstr /c:\"_id\":2  || goto :error
-C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/last?token=%token% --stderr - | findstr /c:\"_id\":2  || goto :error
+C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/%groupid%/last?token=%token% --stderr - | findstr /c:\"_id\":2  || goto :error
+C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/%groupid%/last?token=%token% --stderr - | findstr /c:\"_id\":2  || goto :error
 
 echo db.data.insert({"_id":3}) | %mongo_exe% %database_name%  || goto :error
-C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/last?token=%token% --stderr - | findstr /c:\"_id\":3  || goto :error
+C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/%groupid%/last?token=%token% --stderr - | findstr /c:\"_id\":3  || goto :error
 
 echo db.data.insert({"_id":4}) | %mongo_exe% %database_name%  || goto :error
 
-C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/next?token=%token% --stderr - | findstr /c:\"_id\":4  || goto :error
-C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/last?token=%token% --stderr - | findstr /c:\"_id\":4  || goto :error
+C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/%groupid%/next?token=%token% --stderr - | findstr /c:\"_id\":4  || goto :error
+C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/%groupid%/last?token=%token% --stderr - | findstr /c:\"_id\":4  || goto :error
 
 
 goto :clean
@@ -37,4 +39,5 @@ exit /b 1
 :clean
 Taskkill /IM "%short_name%" /F
 echo db.dropDatabase() | %mongo_exe% %database_name%
-del /f token
\ No newline at end of file
+del /f token
+del /f groupid
\ No newline at end of file

tests/automatic/broker/get_next/check_linux.sh

@@ -22,9 +22,9 @@ $1 -config settings.json &
 sleep 0.3
 brokerid=`echo $!`
 
+groupid=`curl -d '' --silent 127.0.0.1:5005/creategroup`
 
+curl -v  --silent 127.0.0.1:5005/database/data/${groupid}/next?token=$token --stderr - | grep '"_id":1'
+curl -v  --silent 127.0.0.1:5005/database/data/${groupid}/next?token=$token --stderr - | grep '"_id":2'
 
-curl -v  --silent 127.0.0.1:5005/database/data/next?token=$token --stderr - | grep '"_id":1'
-curl -v  --silent 127.0.0.1:5005/database/data/next?token=$token --stderr - | grep '"_id":2'
-
-curl -v  --silent 127.0.0.1:5005/database/data/next?token=$token --stderr - | grep "not found"
+curl -v  --silent 127.0.0.1:5005/database/data/${groupid}/next?token=$token --stderr - | grep "not found"

tests/automatic/broker/get_next/check_windows.bat

@@ -11,14 +11,17 @@ set short_name="%~nx1"
 set /P token=< token
 
 
+curl -d '' --silent 127.0.0.1:5005/creategroup > groupid
+set /P groupid=< groupid
+
 
 start /B "" "%full_name%" -config settings.json
 
 ping 1.0.0.0 -n 1 -w 100 > nul
 
-C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/next?token=%token% --stderr - | findstr /c:\"_id\":1  || goto :error
-C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/next?token=%token% --stderr - | findstr /c:\"_id\":2  || goto :error
-C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/next?token=%token% --stderr - | findstr  /c:"not found"  || goto :error
+C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/%groupid%/next?token=%token% --stderr - | findstr /c:\"_id\":1  || goto :error
+C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/%groupid%/next?token=%token% --stderr - | findstr /c:\"_id\":2  || goto :error
+C:\Curl\curl.exe -v  --silent 127.0.0.1:5005/database/data/%groupid%/next?token=%token% --stderr - | findstr  /c:"not found"  || goto :error
 
 goto :clean
 
@@ -29,4 +32,5 @@ exit /b 1
 :clean
 Taskkill /IM "%short_name%" /F
 echo db.dropDatabase() | %mongo_exe% %database_name%
-del /f token
\ No newline at end of file
+del /f token
+del /f groupid
\ No newline at end of file