Subscription work, adding openssl as a feature

3rd-party/3rd-party.iml

@@ -5,16 +5,5 @@
     <content url="file://$MODULE_DIR$" />
     <orderEntry type="inheritedJdk" />
     <orderEntry type="sourceFolder" forTests="false" />
-    <orderEntry type="module-library">
-      <library name="PHP Runtime" type="php">
-        <CLASSES>
-          <root url="jar://$APPLICATION_PLUGINS_DIR$/php/lib/php.jar!/stubs/standard" />
-        </CLASSES>
-        <JAVADOC />
-        <SOURCES>
-          <root url="jar://$APPLICATION_PLUGINS_DIR$/php/lib/php.jar!/stubs/standard" />
-        </SOURCES>
-      </library>
-    </orderEntry>
   </component>
 </module>

core/Cargo.toml

@@ -11,4 +11,13 @@ rand = "0.3"
 regex = "0.2.1"
 lazy_static = "0.2"
 
-# openssl = "0.9.2"
+# OpenSSL is optional due to problems building and installing it on some platforms.
+openssl = { version = "0.9.9", optional = true }
+
+[features]
+# The default package is no crypto.
+default = []
+
+# Crypto relies on openssl which is a bit of a pain to use on some platforms. Build without if you have trouble with it
+# or simply don't need the functionality.
+crypto = ["openssl"]

core/src/crypto/mod.rs

@@ -0,0 +1,8 @@
+//! Contains the wrappers for crypto for OPC UA.
+//!
+//! This file is an optional component of the stack. If it isn't compiled in, then the OPC UA
+//! impl will not support encryption, decryption, signing or verification.
+
+// TODO
+
+use openssl::*;

docs/docs.iml

@@ -5,16 +5,5 @@
     <content url="file://$MODULE_DIR$" />
     <orderEntry type="inheritedJdk" />
     <orderEntry type="sourceFolder" forTests="false" />
-    <orderEntry type="module-library">
-      <library name="PHP Runtime" type="php">
-        <CLASSES>
-          <root url="jar://$APPLICATION_PLUGINS_DIR$/php/lib/php.jar!/stubs/standard" />
-        </CLASSES>
-        <JAVADOC />
-        <SOURCES>
-          <root url="jar://$APPLICATION_PLUGINS_DIR$/php/lib/php.jar!/stubs/standard" />
-        </SOURCES>
-      </library>
-    </orderEntry>
   </component>
 </module>

server/src/comms/tcp_transport.rs

@@ -26,7 +26,7 @@ const SEND_BUFFER_SIZE: usize = 1024 * 64;
 const MAX_MESSAGE_SIZE: usize = 1024 * 64;
 
 // Rate at which subscriptions are serviced
-const SUBSCRIPTION_TIMER_RATE: i64 = 5000;
+const SUBSCRIPTION_TIMER_RATE: i64 = 500;
 
 #[derive(Clone, Debug, PartialEq)]
 pub enum TransportState {

server/src/services/monitored_item.rs

@@ -42,11 +42,55 @@ impl MonitoredItemService {
         Ok(SupportedMessage::CreateMonitoredItemsResponse(response))
     }
 
-    pub fn modify_monitored_items(&self, _: &mut ServerState, _: &mut SessionState, _: ModifyMonitoredItemsRequest) -> Result<SupportedMessage, &'static StatusCode> {
-        unimplemented!();
+    pub fn modify_monitored_items(&self, _: &mut ServerState, session_state: &mut SessionState, request: ModifyMonitoredItemsRequest) -> Result<SupportedMessage, &'static StatusCode> {
+        let mut service_status = &GOOD;
+        let results = if let Some(ref items_to_modify) = request.items_to_modify {
+            // Find subscription and add items to it
+            let mut subscriptions = session_state.subscriptions.lock().unwrap();
+            let subscription_id = request.subscription_id;
+            if let Some(mut subscription) = subscriptions.get_mut(&subscription_id) {
+                Some(subscription.modify_monitored_items(items_to_modify))
+            } else {
+                // No matching subscription
+                service_status = &BAD_SUBSCRIPTION_ID_INVALID;
+                None
+            }
+        } else {
+            // No items to modify so nothing to do
+            service_status = &BAD_NOTHING_TO_DO;
+            None
+        };
+        let response = ModifyMonitoredItemsResponse {
+            response_header: ResponseHeader::new_service_result(&DateTime::now(), &request.request_header, service_status),
+            results: results,
+            diagnostic_infos: None
+        };
+        Ok(SupportedMessage::ModifyMonitoredItemsResponse(response))
     }
 
-    pub fn delete_monitored_items(&self, _: &mut ServerState, _: &mut SessionState, _: DeleteMonitoredItemsRequest) -> Result<SupportedMessage, &'static StatusCode> {
-        unimplemented!();
+    pub fn delete_monitored_items(&self, _: &mut ServerState, session_state: &mut SessionState, request: DeleteMonitoredItemsRequest) -> Result<SupportedMessage, &'static StatusCode> {
+        let mut service_status = &GOOD;
+        let results = if let Some(ref items_to_delete) = request.monitored_item_ids {
+            // Find subscription and add items to it
+            let mut subscriptions = session_state.subscriptions.lock().unwrap();
+            let subscription_id = request.subscription_id;
+            if let Some(mut subscription) = subscriptions.get_mut(&subscription_id) {
+                Some(subscription.delete_monitored_items(items_to_delete))
+            } else {
+                // No matching subscription
+                service_status = &BAD_SUBSCRIPTION_ID_INVALID;
+                None
+            }
+        } else {
+            // No items to modify so nothing to do
+            service_status = &BAD_NOTHING_TO_DO;
+            None
+        };
+        let response = DeleteMonitoredItemsResponse {
+            response_header: ResponseHeader::new_service_result(&DateTime::now(), &request.request_header, service_status),
+            results: results,
+            diagnostic_infos: None
+        };
+        Ok(SupportedMessage::DeleteMonitoredItemsResponse(response))
     }
 }

server/src/session.rs

@@ -62,7 +62,7 @@ impl SessionState {
                 if subscription.state == SubscriptionState::Closed {
                     dead_subscriptions.push(*subscription_id);
                 } else {
-                    let (notification_message, publish_request_action) = subscription.tick(address_space, &publish_request, &now);
+                    let (notification_message, publish_request_action, subscription_ack_results) = subscription.tick(address_space, &publish_request, &now);
                     if let Some(notification_message) = notification_message {
                         let publish_response = PublishResponse {
                             response_header: ResponseHeader::new_notification_response(&DateTime::now(), &GOOD),
@@ -97,12 +97,7 @@ impl SessionState {
             self.publish_request_queue.push(publish_request.unwrap());
         }
 
-
-        if result.is_empty() {
-            None
-        } else {
-            Some(result)
-        }
+        if result.is_empty() { None } else { Some(result) }
     }
 
 
@@ -117,10 +112,12 @@ impl SessionState {
             } else {
                 MAX_REQUEST_TIMEOUT
             };
-            let timeout = time::Duration::milliseconds(timeout);
+            let timeout_d = time::Duration::milliseconds(timeout);
 
             // The request has timed out if the timestamp plus hint exceeds the input time
-            if timestamp + timeout >= *now {
+            let expiration_time = timestamp + timeout_d;
+            if *now >= expiration_time {
+                debug!("Publish request {} has expired - timestamp = {:?}, expiration hint = {}, expiration time = {:?}, time now = {:?}, ", r.request_header.request_handle, timestamp, timeout, expiration_time, now);
                 let now = DateTime::from_chrono(now);
                 let response = PublishResponse {
                     response_header: ResponseHeader::new_service_result(&now, &r.request_header, &BAD_REQUEST_TIMEOUT),