sca: use new server_id column

* bump required version table to 2

Fix #782

modules/sca/sca_db.c

@@ -36,6 +36,7 @@ const str SCA_DB_TO_TAG_COL_NAME = STR_STATIC_INIT("to_tag");
 const str SCA_DB_RECORD_ROUTE_COL_NAME = STR_STATIC_INIT("record_route");
 const str SCA_DB_NOTIFY_CSEQ_COL_NAME = STR_STATIC_INIT("notify_cseq");
 const str SCA_DB_SUBSCRIBE_CSEQ_COL_NAME = STR_STATIC_INIT("subscribe_cseq");
+const str SCA_DB_SERVER_ID_COL_NAME = STR_STATIC_INIT("server_id");
 
 void sca_db_subscriptions_get_value_for_column(int column, db_val_t *row_values,
 		void *column_value)
@@ -63,6 +64,7 @@ void sca_db_subscriptions_get_value_for_column(int column, db_val_t *row_values,
 	case SCA_DB_SUBS_STATE_COL:
 	case SCA_DB_SUBS_NOTIFY_CSEQ_COL:
 	case SCA_DB_SUBS_SUBSCRIBE_CSEQ_COL:
+	case SCA_DB_SUBS_SERVER_ID_COL:
 		*((int *) column_value) = row_values[column].val.int_val;
 		break;
 
@@ -112,6 +114,7 @@ void sca_db_subscriptions_set_value_for_column(int column, db_val_t *row_values,
 	case SCA_DB_SUBS_STATE_COL:
 	case SCA_DB_SUBS_NOTIFY_CSEQ_COL:
 	case SCA_DB_SUBS_SUBSCRIBE_CSEQ_COL:
+	case SCA_DB_SUBS_SERVER_ID_COL:
 		row_values[column].val.int_val = *((int *) column_value);
 		row_values[column].type = DB1_INT;
 		row_values[column].nul = 0;
@@ -129,6 +132,7 @@ str **sca_db_subscriptions_columns(void)
 			(str *) &SCA_DB_RECORD_ROUTE_COL_NAME,
 			(str *) &SCA_DB_NOTIFY_CSEQ_COL_NAME,
 			(str *) &SCA_DB_SUBSCRIBE_CSEQ_COL_NAME,
+			(str *) &SCA_DB_SERVER_ID_COL_NAME,
 			NULL};
 
 	return (subs_columns);

modules/sca/sca_db.h

@@ -22,9 +22,9 @@
 
 #include "../../lib/srdb1/db.h"
 
-#define SCA_DB_SUBSCRIPTIONS_TABLE_VERSION	1
+#define SCA_DB_SUBSCRIPTIONS_TABLE_VERSION	2
 
-#define SCA_DB_SUBSCRIPTIONS_NUM_COLUMNS	12
+#define SCA_DB_SUBSCRIPTIONS_NUM_COLUMNS	13
 
 #define SCA_DB_DEFAULT_FETCH_ROW_COUNT		1000
 
@@ -41,6 +41,7 @@ enum {
 	SCA_DB_SUBS_RECORD_ROUTE_COL,
 	SCA_DB_SUBS_NOTIFY_CSEQ_COL,
 	SCA_DB_SUBS_SUBSCRIBE_CSEQ_COL,
+	SCA_DB_SUBS_SERVER_ID_COL,
 
 	SCA_DB_SUBS_BOUNDARY,
 };
@@ -78,6 +79,7 @@ extern const str SCA_DB_TO_TAG_COL_NAME;
 extern const str SCA_DB_RECORD_ROUTE_COL_NAME;
 extern const str SCA_DB_NOTIFY_CSEQ_COL_NAME;
 extern const str SCA_DB_SUBSCRIBE_CSEQ_COL_NAME;
+extern const str SCA_DB_SERVER_ID_COL_NAME;
 
 str **sca_db_subscriptions_columns(void);
 void sca_db_subscriptions_get_value_for_column(int, db_val_t *, void *);

modules/sca/sca_subscribe.c

@@ -183,6 +183,8 @@ int sca_subscription_from_db_row_values(db_val_t *values, sca_subscription *sub)
 			values, &sub->dialog.notify_cseq);
 	sca_db_subscriptions_get_value_for_column(SCA_DB_SUBS_SUBSCRIBE_CSEQ_COL,
 			values, &sub->dialog.subscribe_cseq);
+	sca_db_subscriptions_get_value_for_column(SCA_DB_SUBS_SERVER_ID_COL,
+			values, &sub->server_id);
 
 	return (0);
 }
@@ -221,6 +223,8 @@ int sca_subscription_to_db_row_values(sca_subscription *sub, db_val_t *values)
 			values, &notify_cseq);
 	sca_db_subscriptions_set_value_for_column(SCA_DB_SUBS_SUBSCRIBE_CSEQ_COL,
 			values, &subscribe_cseq);
+	sca_db_subscriptions_set_value_for_column(SCA_DB_SUBS_SERVER_ID_COL,
+			values, &sub->server_id);
 
 	return (0);
 }
@@ -228,6 +232,9 @@ int sca_subscription_to_db_row_values(sca_subscription *sub, db_val_t *values)
 int sca_subscriptions_restore_from_db(sca_mod *scam)
 {
 	db1_con_t *db_con;
+	db_key_t query_columns[1];
+	db_val_t query_values[1];
+	db_op_t query_ops[1];
 	db_key_t result_columns[SCA_DB_SUBSCRIPTIONS_NUM_COLUMNS];
 	db1_res_t *result = NULL;
 	db_row_t *rows = NULL;
@@ -238,6 +245,7 @@ int sca_subscriptions_restore_from_db(sca_mod *scam)
 	int num_rows;
 	int i;
 	int idx;
+	int q_count = 0;
 	int rc = -1;
 	time_t now = time(NULL);
 
@@ -261,8 +269,13 @@ int sca_subscriptions_restore_from_db(sca_mod *scam)
 		result_columns[i] = column_names[i];
 	}
 
+	query_columns[q_count] = (str *) &SCA_DB_SERVER_ID_COL_NAME;
+	query_ops[q_count] = OP_EQ;
+	SCA_DB_BIND_INT_VALUE(server_id, &SCA_DB_SERVER_ID_COL_NAME, query_columns,
+		query_values, q_count);
+
 	rc = db_fetch_query(scam->db_api, SCA_DB_DEFAULT_FETCH_ROW_COUNT, db_con,
-			NULL, NULL, NULL, result_columns, 0,
+			query_columns, query_ops, query_values, result_columns, q_count,
 			SCA_DB_SUBSCRIPTIONS_NUM_COLUMNS, 0, &result);
 	switch (rc) {
 	default:
@@ -439,14 +452,20 @@ static int sca_subscription_db_insert_subscriber(db1_con_t *db_con,
 
 int sca_subscription_db_delete_expired(db1_con_t *db_con)
 {
-	db_key_t delete_columns[1];
-	db_val_t delete_values[1];
-	db_op_t delete_ops[1];
+	db_key_t delete_columns[2];
+	db_val_t delete_values[2];
+	db_op_t delete_ops[2];
 	time_t now = time(NULL);
 	int kv_count = 0;
 
-	delete_columns[0] = (str *) &SCA_DB_EXPIRES_COL_NAME;
-	delete_ops[0] = OP_LT;
+	delete_columns[kv_count] = (str *) &SCA_DB_SERVER_ID_COL_NAME;
+	delete_ops[kv_count] = OP_EQ;
+
+	SCA_DB_BIND_INT_VALUE(server_id, &SCA_DB_SERVER_ID_COL_NAME, delete_columns,
+			delete_values, kv_count);
+
+	delete_columns[kv_count] = (str *) &SCA_DB_EXPIRES_COL_NAME;
+	delete_ops[kv_count] = OP_LT;
 
 	SCA_DB_BIND_INT_VALUE(now, &SCA_DB_EXPIRES_COL_NAME, delete_columns,
 			delete_values, kv_count);
@@ -657,6 +676,8 @@ sca_subscription *sca_subscription_create(str *aor, int event, str *subscriber,
 	sub->dialog.to_tag.s = sub->dialog.id.s + call_id->len + from_tag->len;
 	sub->dialog.to_tag.len = to_tag->len;
 
+	sub->server_id = server_id;
+
 	return (sub);
 
 	error: if (sub != NULL) {
@@ -706,7 +727,7 @@ void sca_subscription_print(void *value)
 
 	LM_DBG("%.*s %s (%d) %.*s, expires: %ld, index: %d, "
 			"dialog %.*s;%.*s;%.*s, record_route: %.*s, "
-			"notify_cseq: %d, subscribe_cseq: %d\n",
+			"notify_cseq: %d, subscribe_cseq: %d, server_id: %d\n",
 			STR_FMT(&sub->target_aor),
 			sca_event_name_from_type(sub->event),
 			sub->event,
@@ -718,7 +739,8 @@ void sca_subscription_print(void *value)
 			SCA_STR_EMPTY(&sub->rr) ? 4 : sub->rr.len,
 			SCA_STR_EMPTY(&sub->rr) ? "null" : sub->rr.s,
 			sub->dialog.notify_cseq,
-			sub->dialog.subscribe_cseq);
+			sub->dialog.subscribe_cseq,
+			sub->server_id);
 }
 
 int sca_subscription_save_unsafe(sca_mod *scam, sca_subscription *sub,
@@ -1094,6 +1116,7 @@ int sca_subscription_from_request(sca_mod *scam, sip_msg_t *msg, int event_type,
 
 	req_sub->dialog.subscribe_cseq = 0;
 	req_sub->dialog.notify_cseq = 0;
+	req_sub->server_id = server_id;
 
 	free_to_params(&tmp_to);
 

modules/sca/sca_subscribe.h

@@ -48,6 +48,7 @@ struct _sca_subscription {
 	str rr; // Record-Route header values
 
 	int db_cmd_flag; // track whether to INSERT or UPDATE
+	int server_id; // server
 };
 typedef struct _sca_subscription sca_subscription;