Source Subscription deletion #136
Conversation
ruokun-niu
requested review from
amansinghoriginal,
danielgerlag,
agentofreality and
Copilot
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| key: &str, | ||
| metadata: Option<HashMap<String, String>>, | ||
| ) -> Result<(), Box<dyn std::error::Error>> { | ||
| let addr = "https://127.0.0.1".to_string(); |
Check notice
Code scanning / devskim
Accessing localhost could indicate debug code, or could hinder scaling. Note
There was a problem hiding this comment.
Copilot reviewed 6 out of 8 changed files in this pull request and generated no comments.
Files not reviewed (2)
- query-container/query-host/src/query_worker.rs: Evaluated as low risk
- sources/shared/query-api/src/api.rs: Evaluated as low risk
Comments suppressed due to low confidence (1)
query-container/query-host/src/models.rs:178
- The error message 'Failed to unsubscribe' could be more descriptive. Consider including the query ID or subscription ID in the message.
UnsubscribeFailed(String),
|
|
||
| let resp = match self | ||
| .client | ||
| .post(format!("http://{}/unsubscription", app_id)) |
There was a problem hiding this comment.
To remain "RESTful", I think this should be a DELETE verb on the subscription/{queryNodeId}/{queryId} route.
| state: &Arc<AppState>, | ||
| ) -> Result<(), axum::http::Response<axum::body::Body>> { | ||
| let publisher = &state.publisher; | ||
| let unsubscription_event = api::UnsubscriptionEvent { |
There was a problem hiding this comment.
Could this just be a ControlEvent with an op of d instead of i on the SourceSubscription table?
| @@ -391,6 +391,31 @@ async fn process_changes( | |||
| } else { | |||
| // TODO - supprt other ops on SourceSubscriptions | |||
| } | |||
| } else if change["payload"]["source"]["table"] == "SourceUnsubscription" { | |||
There was a problem hiding this comment.
Could we refactor these deep nested if else statements into a match statement?
Description
Currently, when a continuous query is deleted, the subscription info in the change router is still persisted. This PR focuses on the deletion of subscription when a continuous query is deleted. Specifically, this is achieved by the following steps:
DELETEcommand is received in the query worker, the worker will iterate through all of the sources that the query is subscribed to, and then invoke theunsubscriptionmethod in the sources' query-api (this is achieved through dapr service invocation).unsubscriptionmethod is invoked, the query api will publish the following message to the Source's change router:{"payload":{"queryId":"<query-id>","queryNodeId":"<query-node-id>","source":{"db":"Drasi","table":"SourceUnsubscription"}}}This message has a similar format as the SourceSubscription message
Type of change
Fixes: #issue_number