Update for new version of simd-json and value trait

Cargo.toml

@@ -83,8 +83,8 @@ rand = "0.8.5"
 regex = "1.9"
 serde = { version = "1", features = ["derive"] }
 serde_yaml = "0.9"
-simd-json = { version = "0.12", features = ["known-key"] }
-simd-json-derive = "0.12"
+simd-json = { version = "0.13", features = ["known-key"] }
+simd-json-derive = "0.13"
 socket2 = { version = "0.5", features = ["all"] }
 tremor-common = { path = "tremor-common" }
 tremor-config = { path = "tremor-config" }
@@ -95,7 +95,7 @@ tremor-script = { path = "tremor-script" }
 tremor-value = { path = "tremor-value" }
 tremor-interceptor = { path = "tremor-interceptor" }
 url = "2.4"
-value-trait = "0.6"
+value-trait = "0.8"
 
 
 # blaster / blackhole# codecs

depricated/sink/amqp.rs

@@ -156,7 +156,7 @@ impl Sink for Amqp {
                         for (key, val) in
                             headers.iter().filter_map(|(k, v)| Some((k, v.as_array()?)))
                         {
-                            for ele in val.iter().filter_map(value_trait::ValueAccess::as_str) {
+                            for ele in val.iter().filter_map(value_trait::ValueAsScalar::as_str) {
                                 key_val.push((key, ele));
                             }
                         }

depricated/sink/nats.rs

@@ -200,7 +200,7 @@ impl Sink for Nats {
                         for (key, val) in
                             headers.iter().filter_map(|(k, v)| Some((k, v.as_array()?)))
                         {
-                            for ele in val.iter().filter_map(value_trait::ValueAccess::as_str) {
+                            for ele in val.iter().filter_map(value_trait::ValueAsScalar::as_str) {
                                 key_val.push((key, ele));
                             }
                         }

src/connectors.rs

@@ -57,7 +57,7 @@ use tremor_value::Value;
 use utils::reconnect::{Attempt, ConnectionLostNotifier, ReconnectRuntime};
 /// quiescence stuff
 pub(crate) use utils::{metrics, reconnect};
-use value_trait::{Builder, Mutable, ValueAccess};
+use value_trait::prelude::*;
 
 /// Accept timeout
 pub(crate) const ACCEPT_TIMEOUT: Duration = Duration::from_millis(100);

src/connectors/google.rs

@@ -62,7 +62,7 @@ impl TokenSrc {
     }
 }
 
-pub trait TokenProvider: Clone + Send + From<TokenSrc> {
+pub trait TokenProvider: Clone + Send + Sync + From<TokenSrc> {
     fn get_token(&mut self) -> ::std::result::Result<Arc<String>, Status>;
 }
 

src/connectors/impls/clickhouse/conversion.rs

@@ -21,9 +21,9 @@ use std::{
 use chrono_tz::Tz;
 pub(super) use clickhouse_rs::types::Value as CValue;
 use either::Either;
-use simd_json::{Value, ValueAccess};
 use tremor_value::Value as TValue;
 use uuid::Uuid;
+use value_trait::prelude::*;
 
 use super::DummySqlType;
 use crate::errors::{Error, ErrorKind, Result};
@@ -46,7 +46,7 @@ pub(super) fn convert_value(
             // We don't check that all elements of the array have the same type.
             // Instead, we check that every element can be converted to the expected
             // array type.
-            wrap_getter_error(context, ValueAccess::as_array)?
+            wrap_getter_error(context, ValueAsContainer::as_array)?
                 .iter()
                 .map(|value| convert_value(column_name, value, expected_inner_type))
                 .collect::<Result<Vec<_>>>()
@@ -72,23 +72,23 @@ pub(super) fn convert_value(
             Ok(CValue::Nullable(Either::Right(Box::new(inner_value))))
         }
 
-        DummySqlType::UInt8 => get_and_wrap(context, ValueAccess::as_u8, CValue::UInt8),
+        DummySqlType::UInt8 => get_and_wrap(context, ValueAsScalar::as_u8, CValue::UInt8),
 
-        DummySqlType::UInt16 => get_and_wrap(context, ValueAccess::as_u16, CValue::UInt16),
+        DummySqlType::UInt16 => get_and_wrap(context, ValueAsScalar::as_u16, CValue::UInt16),
 
-        DummySqlType::UInt32 => get_and_wrap(context, ValueAccess::as_u32, CValue::UInt32),
+        DummySqlType::UInt32 => get_and_wrap(context, ValueAsScalar::as_u32, CValue::UInt32),
 
-        DummySqlType::UInt64 => get_and_wrap(context, ValueAccess::as_u64, CValue::UInt64),
+        DummySqlType::UInt64 => get_and_wrap(context, ValueAsScalar::as_u64, CValue::UInt64),
 
-        DummySqlType::Int8 => get_and_wrap(context, ValueAccess::as_i8, CValue::Int8),
+        DummySqlType::Int8 => get_and_wrap(context, ValueAsScalar::as_i8, CValue::Int8),
 
-        DummySqlType::Int16 => get_and_wrap(context, ValueAccess::as_i16, CValue::Int16),
+        DummySqlType::Int16 => get_and_wrap(context, ValueAsScalar::as_i16, CValue::Int16),
 
-        DummySqlType::Int32 => get_and_wrap(context, ValueAccess::as_i32, CValue::Int32),
+        DummySqlType::Int32 => get_and_wrap(context, ValueAsScalar::as_i32, CValue::Int32),
 
-        DummySqlType::Int64 => get_and_wrap(context, ValueAccess::as_i64, CValue::Int64),
+        DummySqlType::Int64 => get_and_wrap(context, ValueAsScalar::as_i64, CValue::Int64),
 
-        DummySqlType::String => get_and_wrap(context, ValueAccess::as_str, |s| {
+        DummySqlType::String => get_and_wrap(context, ValueAsScalar::as_str, |s| {
             CValue::String(Arc::new(s.as_bytes().to_vec()))
         }),
 
@@ -116,25 +116,31 @@ pub(super) fn convert_value(
             ErrorKind::MalformedUuid,
         ),
 
-        DummySqlType::DateTime => get_and_wrap(context, ValueAccess::as_u32, |timestamp| {
+        DummySqlType::DateTime => get_and_wrap(context, ValueAsScalar::as_u32, |timestamp| {
             CValue::DateTime(timestamp, UTC)
         }),
 
-        DummySqlType::DateTime64Secs => get_and_wrap(context, ValueAccess::as_i64, |timestamp| {
+        DummySqlType::DateTime64Secs => get_and_wrap(context, ValueAsScalar::as_i64, |timestamp| {
             CValue::DateTime64(timestamp, (0, UTC))
         }),
 
-        DummySqlType::DateTime64Millis => get_and_wrap(context, ValueAccess::as_i64, |timestamp| {
-            CValue::DateTime64(timestamp, (3, UTC))
-        }),
+        DummySqlType::DateTime64Millis => {
+            get_and_wrap(context, ValueAsScalar::as_i64, |timestamp| {
+                CValue::DateTime64(timestamp, (3, UTC))
+            })
+        }
 
-        DummySqlType::DateTime64Micros => get_and_wrap(context, ValueAccess::as_i64, |timestamp| {
-            CValue::DateTime64(timestamp, (6, UTC))
-        }),
+        DummySqlType::DateTime64Micros => {
+            get_and_wrap(context, ValueAsScalar::as_i64, |timestamp| {
+                CValue::DateTime64(timestamp, (6, UTC))
+            })
+        }
 
-        DummySqlType::DateTime64Nanos => get_and_wrap(context, ValueAccess::as_i64, |timestamp| {
-            CValue::DateTime64(timestamp, (9, UTC))
-        }),
+        DummySqlType::DateTime64Nanos => {
+            get_and_wrap(context, ValueAsScalar::as_i64, |timestamp| {
+                CValue::DateTime64(timestamp, (9, UTC))
+            })
+        }
     }
 }
 

src/connectors/impls/crononome.rs

@@ -119,6 +119,7 @@ use crate::{connectors::prelude::*, errors::err_connector_def, system::KillSwitc
 use handler::{ChronomicQueue, CronEntryInt};
 use serde_yaml::Value as YamlValue;
 use tremor_common::time::nanotime;
+use value_trait::ValueBuilder;
 
 #[derive(Deserialize, Debug, Clone)]
 #[serde(deny_unknown_fields)]

src/connectors/impls/elastic.rs

@@ -258,7 +258,6 @@ use tokio::task;
 use tremor_common::time::nanotime;
 use tremor_value::utils::sorted_serialize;
 use tremor_value::value::StaticValue;
-use value_trait::Mutable;
 
 #[derive(Deserialize, Debug, Clone)]
 #[serde(deny_unknown_fields)]
@@ -746,7 +745,10 @@ async fn handle_response(
 ) -> Result<()> {
     let correlation_values = event.correlation_metas();
     let payload_iter = event.value_iter();
-    if let Some(items) = response.get_mut("items").and_then(Mutable::as_array_mut) {
+    if let Some(items) = response
+        .get_mut("items")
+        .and_then(ValueAsMutContainer::as_array_mut)
+    {
         for ((mut item, correlation), payload) in items
             .drain(..)
             .zip(correlation_values.into_iter())

src/connectors/impls/exit.rs

@@ -94,7 +94,6 @@
 use crate::connectors::prelude::*;
 use crate::system::{KillSwitch, ShutdownMode};
 use std::time::Duration;
-use value_trait::ValueAccess;
 
 #[derive(Deserialize, Debug, Clone)]
 #[serde(deny_unknown_fields)]

src/connectors/impls/gcl/writer/meta.rs

@@ -12,14 +12,12 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::errors::Result;
+use crate::connectors::prelude::*;
 use googapis::google::logging::{
     r#type::HttpRequest,
     v2::{LogEntryOperation, LogEntrySourceLocation},
 };
 use prost_types::Timestamp;
-use tremor_value::Value;
-use value_trait::ValueAccess;
 
 pub(crate) fn get_or_default(meta: Option<&Value>, key: &str) -> String {
     meta.get_str(key).unwrap_or_default().to_string()

src/connectors/impls/gpubsub/producer.rs

@@ -27,7 +27,6 @@ use tonic::Code;
 use tremor_common::url::HttpsDefaults;
 use tremor_pipeline::Event;
 use tremor_value::Value;
-use value_trait::ValueAccess;
 
 #[derive(Deserialize, Clone)]
 #[serde(deny_unknown_fields)]

src/connectors/impls/http/meta.rs

@@ -25,7 +25,6 @@ use http::{
 use hyper::{header::HeaderValue, Body, Method, Request, Response};
 use mime::Mime;
 use tremor_value::Value;
-use value_trait::{Builder, ValueAccess};
 
 /// Utility for building an HTTP request from a possibly batched event
 /// and some configuration values

src/connectors/impls/http/server.rs

@@ -38,7 +38,6 @@ use hyper::{
     service::{make_service_fn, service_fn},
     Body, Request,
 };
-use simd_json::ValueAccess;
 use std::{
     convert::Infallible,
     net::{SocketAddr, ToSocketAddrs},

src/connectors/impls/otel/common.rs

@@ -14,18 +14,14 @@
 
 #![allow(dead_code)]
 
+use crate::connectors::prelude::*;
 use crate::connectors::utils::pb;
-use crate::errors::Result;
-use simd_json::Builder;
-use tremor_common::url;
 use tremor_otelapis::opentelemetry::proto::common::v1::{
     any_value, AnyValue, ArrayValue, InstrumentationLibrary, KeyValue, KeyValueList, StringKeyValue,
 };
-use tremor_value::{literal, StaticNode, Value};
-use value_trait::ValueAccess;
 
 pub(crate) struct OtelDefaults;
-impl url::Defaults for OtelDefaults {
+impl Defaults for OtelDefaults {
     // We do add the port here since it's different from http's default
 
     const SCHEME: &'static str = "https";

src/connectors/impls/otel/id.rs

@@ -14,12 +14,10 @@
 
 #![allow(dead_code)]
 
-use crate::errors::Result;
+use crate::connectors::prelude::*;
 use ::rand::Rng;
-use simd_json::ValueAccess;
 use std::fmt::Write;
 use tremor_common::rand;
-use tremor_value::Value;
 
 pub(crate) fn random_span_id_bytes(ingest_ns_seed: u64) -> Vec<u8> {
     let mut rng = tremor_common::rand::make_prng(ingest_ns_seed);

src/connectors/impls/otel/resource.rs

@@ -15,11 +15,9 @@
 #![allow(dead_code)]
 
 use super::common;
+use crate::connectors::prelude::*;
 use crate::connectors::utils::pb;
-use crate::errors::Result;
-use simd_json::ValueAccess;
 use tremor_otelapis::opentelemetry::proto::resource::v1::Resource;
-use tremor_value::{literal, Value};
 
 pub(crate) fn resource_to_json(pb: Resource) -> Value<'static> {
     literal!({

src/connectors/impls/otel/trace.rs

@@ -19,13 +19,10 @@ use super::{
     id,
     resource::{self, resource_to_pb},
 };
+use crate::connectors::prelude::*;
 use crate::connectors::utils::pb::{
     maybe_int_to_pbi32, maybe_int_to_pbu32, maybe_int_to_pbu64, maybe_string_to_pb,
 };
-use crate::errors::Result;
-use simd_json::Mutable;
-use tremor_value::literal;
-
 use tremor_otelapis::opentelemetry::proto::{
     collector::trace::v1::ExportTraceServiceRequest,
     trace::v1::{
@@ -34,9 +31,6 @@ use tremor_otelapis::opentelemetry::proto::{
     },
 };
 
-use tremor_value::Value;
-use value_trait::ValueAccess;
-
 #[allow(deprecated)]
 pub(crate) fn status_to_json<'event>(data: Option<Status>) -> Value<'event> {
     data.map_or_else(

src/connectors/impls/tcp/server.rs

@@ -29,7 +29,6 @@ use crate::{
     errors::err_connector_def,
 };
 use rustls::ServerConfig;
-use simd_json::ValueAccess;
 use std::sync::{atomic::AtomicBool, Arc};
 use tokio::{task::JoinHandle, time::timeout};
 use tokio_rustls::TlsAcceptor;

src/connectors/impls/ws/server.rs

@@ -23,7 +23,6 @@ use crate::connectors::{
 };
 use futures::StreamExt;
 use rustls::ServerConfig;
-use simd_json::ValueAccess;
 use std::{
     net::SocketAddr,
     sync::{atomic::AtomicBool, Arc},

src/connectors/sink/channel_sink.rs

@@ -38,7 +38,6 @@ use tokio::{
 use tremor_common::{ids::Id, time::nanotime};
 use tremor_pipeline::{CbAction, Event, SignalKind};
 use tremor_value::Value;
-use value_trait::ValueAccess;
 
 /// Behavioral trait for defining if a Channel Sink needs metadata or not
 pub(crate) trait SinkMetaBehaviour: Send + Sync {

src/connectors/source.rs

@@ -17,9 +17,11 @@
 /// A simple source that is fed with `SourceReply` via a channel.
 pub mod channel_source;
 
+use super::{CodecReq, Connectivity};
 use crate::channel::{unbounded, Sender, UnboundedReceiver, UnboundedSender};
 use crate::connectors::{
     metrics::SourceReporter,
+    prelude::*,
     utils::reconnect::{Attempt, ConnectionLostNotifier},
     ConnectorType, Context, Msg, QuiescenceBeacon, StreamDone,
 };
@@ -29,7 +31,6 @@
 use crate::pipeline::InputTarget;
 pub(crate) use channel_source::{ChannelSource, ChannelSourceRuntime};
 use hashbrown::HashSet;
-use simd_json::Mutable;
 use std::collections::{btree_map::Entry, BTreeMap};
 use std::fmt::Display;
 use tokio::task;
@@ -48,9 +49,6 @@
 };
 use tremor_script::{ast::DeployEndpoint, prelude::BaseExpr, EventPayload, ValueAndMeta};
 use tremor_value::{literal, Value};
-use value_trait::Builder;
-
-use super::{CodecReq, Connectivity};
 
 #[derive(Debug)]
 /// Messages a Source can receive
@@ -343,7 +341,7 @@
     /// if the source can not be spawned into a own process
     pub(crate) fn spawn<S>(self, source: S, ctx: SourceContext) -> SourceAddr
     where
-        S: Source + Send + 'static,
+        S: Source + Send + Sync + 'static,
     {
         // We use a unbounded channel for counterflow, while an unbounded channel seems dangerous
         // there is soundness to this.
@@ -855,8 +853,20 @@
     }
 
     /// send a signal to all connected pipelines
-    async fn send_signal(&mut self, signal: Event) -> Result<()> {
-        for (_url, addr) in self.pipelines_out.iter().chain(self.pipelines_err.iter()) {
+    async fn send_signal(&self, signal: Event) -> Result<()> {
+        for (_url, addr) in self
+            .pipelines_out
+            .as_slice()
+            .iter()
+            .chain(self.pipelines_err.as_slice().iter())
+        /* */
+        {
+            addr.send(Box::new(pipeline::Msg::Signal(signal.clone())))
+                .await?;
+        }
+        for (_url, addr) in &self.pipelines_out
+        /*.chain(self.pipelines_err.iter()) */
+        {
             addr.send(Box::new(pipeline::Msg::Signal(signal.clone())))
                 .await?;
         }

src/connectors/tests/elastic.rs

@@ -34,7 +34,7 @@ use tokio::process;
 use tremor_common::ports::IN;
 use tremor_pipeline::{CbAction, Event, EventId};
 use tremor_value::{literal, value::StaticValue};
-use value_trait::{Mutable, Value, ValueAccess};
+use value_trait::prelude::*;
 
 const IMAGE: &str = "elasticsearch";
 const VERSION: &str = "8.6.2";