nu_protocol/pipeline/
pipeline_data.rs

1#[cfg(feature = "os")]
2use crate::process::ExitStatusGuard;
3use crate::{
4    ByteStream, ByteStreamSource, ByteStreamType, CompareTypes, Config, ListStream, OutDest,
5    PipelineMetadata, Range, ShellError, Signals, Span, Type, TypeRelation, Value,
6    ast::{Call, PathMember},
7    engine::{EngineState, Stack},
8    shell_error::{generic::GenericError, io::IoError},
9};
10use std::{
11    borrow::Cow,
12    io::Write,
13    ops::{Deref, DerefMut},
14    panic::Location,
15};
16
17const LINE_ENDING_PATTERN: &[char] = &['\r', '\n'];
18
19/// The foundational abstraction for input and output to commands
20///
21/// This represents either a single Value or a stream of values coming into the command or leaving a command.
22///
23/// A note on implementation:
24///
25/// We've tried a few variations of this structure. Listing these below so we have a record.
26///
27/// * We tried always assuming a stream in Nushell. This was a great 80% solution, but it had some rough edges.
28///   Namely, how do you know the difference between a single string and a list of one string. How do you know
29///   when to flatten the data given to you from a data source into the stream or to keep it as an unflattened
30///   list?
31///
32/// * We tried putting the stream into Value. This had some interesting properties as now commands "just worked
33///   on values", but lead to a few unfortunate issues.
34///
35/// The first is that you can't easily clone Values in a way that felt largely immutable. For example, if
36/// you cloned a Value which contained a stream, and in one variable drained some part of it, then the second
37/// variable would see different values based on what you did to the first.
38///
39/// To make this kind of mutation thread-safe, we would have had to produce a lock for the stream, which in
40/// practice would have meant always locking the stream before reading from it. But more fundamentally, it
41/// felt wrong in practice that observation of a value at runtime could affect other values which happen to
42/// alias the same stream. By separating these, we don't have this effect. Instead, variables could get
43/// concrete list values rather than streams, and be able to view them without non-local effects.
44///
45/// * A balance of the two approaches is what we've landed on: Values are thread-safe to pass, and we can stream
46///   them into any sources. Streams are still available to model the infinite streams approach of original
47///   Nushell.
48#[derive(Debug)]
49pub enum PipelineData {
50    Empty,
51    Value(Value, Option<PipelineMetadata>),
52    ListStream(ListStream, Option<PipelineMetadata>),
53    ByteStream(ByteStream, Option<PipelineMetadata>),
54}
55
56impl PipelineData {
57    pub const fn empty() -> PipelineData {
58        PipelineData::Empty
59    }
60
61    pub fn value(val: Value, metadata: impl Into<Option<PipelineMetadata>>) -> Self {
62        PipelineData::Value(val, metadata.into())
63    }
64
65    pub fn list_stream(stream: ListStream, metadata: impl Into<Option<PipelineMetadata>>) -> Self {
66        PipelineData::ListStream(stream, metadata.into())
67    }
68
69    pub fn byte_stream(stream: ByteStream, metadata: impl Into<Option<PipelineMetadata>>) -> Self {
70        PipelineData::ByteStream(stream, metadata.into())
71    }
72
73    /// Returns a clone of the metadata if it exists.
74    ///
75    /// Note: This performs a deep clone of heap-allocated structures.
76    /// Use [`.metadata_ref()`](Self::metadata_ref), [`.metadata_mut()`](Self::metadata_mut)
77    /// or [`.take_metadata()`](Self::take_metadata) to avoid unnecessary allocations.
78    #[deprecated(
79        since = "0.111.1",
80        note = "Use .metadata_ref(), .metadata_mut() or .take_metadata() instead"
81    )]
82    pub fn metadata(&self) -> Option<PipelineMetadata> {
83        self.metadata_ref().cloned()
84    }
85
86    /// Returns a reference to the metadata if it exists.
87    pub fn metadata_ref(&self) -> Option<&PipelineMetadata> {
88        match self {
89            PipelineData::Empty => None,
90            PipelineData::Value(_, meta)
91            | PipelineData::ListStream(_, meta)
92            | PipelineData::ByteStream(_, meta) => meta.as_ref(),
93        }
94    }
95
96    /// Returns a mutable reference to the metadata if it exists.
97    pub fn metadata_mut(&mut self) -> Option<&mut PipelineMetadata> {
98        match self {
99            PipelineData::Empty => None,
100            PipelineData::Value(_, meta)
101            | PipelineData::ListStream(_, meta)
102            | PipelineData::ByteStream(_, meta) => meta.as_mut(),
103        }
104    }
105
106    /// Take the metadata out of pipeline if it exists.
107    pub fn take_metadata(&mut self) -> Option<PipelineMetadata> {
108        match self {
109            PipelineData::Empty => None,
110            PipelineData::Value(_, meta)
111            | PipelineData::ListStream(_, meta)
112            | PipelineData::ByteStream(_, meta) => meta.take(),
113        }
114    }
115
116    pub fn set_metadata(mut self, metadata: Option<PipelineMetadata>) -> Self {
117        match &mut self {
118            PipelineData::Empty => {}
119            PipelineData::Value(_, meta)
120            | PipelineData::ListStream(_, meta)
121            | PipelineData::ByteStream(_, meta) => *meta = metadata,
122        }
123        self
124    }
125
126    pub fn is_nothing(&self) -> bool {
127        matches!(self, PipelineData::Value(Value::Nothing { .. }, ..))
128            || matches!(self, PipelineData::Empty)
129    }
130
131    /// PipelineData doesn't always have a Span, but we can try!
132    pub fn span(&self) -> Option<Span> {
133        match self {
134            PipelineData::Empty => None,
135            PipelineData::Value(value, ..) => Some(value.span()),
136            PipelineData::ListStream(stream, ..) => Some(stream.span()),
137            PipelineData::ByteStream(stream, ..) => Some(stream.span()),
138        }
139    }
140
141    /// Change the span of the [`PipelineData`].
142    ///
143    /// Returns `Value(Nothing)` with the given span if it was [`PipelineData::empty()`].
144    pub fn with_span(self, span: Span) -> Self {
145        match self {
146            PipelineData::Empty => PipelineData::value(Value::nothing(span), None),
147            PipelineData::Value(value, metadata) => {
148                PipelineData::value(value.with_span(span), metadata)
149            }
150            PipelineData::ListStream(stream, metadata) => {
151                PipelineData::list_stream(stream.with_span(span), metadata)
152            }
153            PipelineData::ByteStream(stream, metadata) => {
154                PipelineData::byte_stream(stream.with_span(span), metadata)
155            }
156        }
157    }
158
159    /// Get a type that is representative of the `PipelineData`.
160    ///
161    /// The type returned here makes no effort to collect a stream, so it may be a different type
162    /// than would be returned by [`Value::get_type()`] on the result of
163    /// [`.into_value()`](Self::into_value).
164    ///
165    /// Specifically, a `ListStream` results in `list<any>` rather than
166    /// the fully complete [`list`](Type::List) type (which would require knowing the contents),
167    /// and a `ByteStream` with [unknown](crate::ByteStreamType::Unknown) type results in
168    /// [`any`](Type::Any) rather than [`string`](Type::String) or [`binary`](Type::Binary).
169    pub fn get_type(&self) -> Type {
170        match self {
171            PipelineData::Empty => Type::Nothing,
172            PipelineData::Value(value, _) => value.get_type(),
173            PipelineData::ListStream(_, _) => Type::list(Type::Any),
174            PipelineData::ByteStream(stream, _) => stream.type_().into(),
175        }
176    }
177
178    pub fn into_value(self, span: Span) -> Result<Value, ShellError> {
179        match self {
180            PipelineData::Empty => Ok(Value::nothing(span)),
181            PipelineData::Value(value, ..) => {
182                if value.span() == Span::unknown() {
183                    Ok(value.with_span(span))
184                } else {
185                    Ok(value)
186                }
187            }
188            PipelineData::ListStream(stream, ..) => stream.into_value(),
189            PipelineData::ByteStream(stream, ..) => stream.into_value(),
190        }
191    }
192
193    /// Converts any `Value` variant that can be represented as a stream into its stream variant.
194    ///
195    /// This means that lists and ranges are converted into list streams, and strings and binary are
196    /// converted into byte streams.
197    ///
198    /// Returns an `Err` with the original stream if the variant couldn't be converted to a stream
199    /// variant. If the variant is already a stream variant, it is returned as-is.
200    pub fn try_into_stream(self, engine_state: &EngineState) -> Result<PipelineData, PipelineData> {
201        let span = self.span().unwrap_or(Span::unknown());
202        match self {
203            PipelineData::ListStream(..) | PipelineData::ByteStream(..) => Ok(self),
204            PipelineData::Value(Value::List { .. } | Value::Range { .. }, ref metadata) => {
205                let metadata = metadata.clone();
206                Ok(PipelineData::list_stream(
207                    ListStream::new(self.into_iter(), span, engine_state.signals().clone()),
208                    metadata,
209                ))
210            }
211            PipelineData::Value(Value::String { val, .. }, metadata) => {
212                Ok(PipelineData::byte_stream(
213                    ByteStream::read_string(val, span, engine_state.signals().clone()),
214                    metadata,
215                ))
216            }
217            PipelineData::Value(Value::Binary { val, .. }, metadata) => {
218                Ok(PipelineData::byte_stream(
219                    ByteStream::read_binary(val, span, engine_state.signals().clone()),
220                    metadata,
221                ))
222            }
223            PipelineData::Value(Value::Custom { val, internal_span }, metadata) => {
224                match val.to_base_value(internal_span) {
225                    Ok(Value::List { vals, .. }) => Ok(PipelineData::list_stream(
226                        ListStream::new(vals.into_iter(), span, engine_state.signals().clone()),
227                        metadata,
228                    )),
229                    Ok(Value::Range { val, .. }) => Ok(PipelineData::list_stream(
230                        ListStream::new(
231                            val.into_range_iter(span, Signals::empty()),
232                            span,
233                            engine_state.signals().clone(),
234                        ),
235                        metadata,
236                    )),
237                    Ok(other) => Err(PipelineData::value(other, metadata)),
238                    Err(_) => Err(PipelineData::Value(
239                        Value::Custom { val, internal_span },
240                        metadata,
241                    )),
242                }
243            }
244            _ => Err(self),
245        }
246    }
247
248    /// Converts this value into a stream when possible, otherwise returns the original value.
249    ///
250    /// This is a convenience wrapper around [`PipelineData::try_into_stream`] for command code
251    /// paths that can operate on both stream and non-stream input without branching.
252    #[must_use]
253    pub fn into_stream_or_original(self, engine_state: &EngineState) -> PipelineData {
254        self.try_into_stream(engine_state)
255            .unwrap_or_else(|original| original)
256    }
257
258    /// Drain and write this [`PipelineData`] to `dest`.
259    ///
260    /// Values are converted to bytes and separated by newlines if this is a `ListStream`.
261    pub fn write_to(self, mut dest: impl Write) -> Result<(), ShellError> {
262        match self {
263            PipelineData::Empty => Ok(()),
264            PipelineData::Value(value, ..) => {
265                let bytes = value_to_bytes(value)?;
266                dest.write_all(&bytes).map_err(|err| {
267                    IoError::new_internal(err, "Could not write PipelineData to dest")
268                })?;
269                dest.flush().map_err(|err| {
270                    IoError::new_internal(err, "Could not flush PipelineData to dest")
271                })?;
272                Ok(())
273            }
274            PipelineData::ListStream(stream, ..) => {
275                for value in stream {
276                    let bytes = value_to_bytes(value)?;
277                    dest.write_all(&bytes).map_err(|err| {
278                        IoError::new_internal(err, "Could not write PipelineData to dest")
279                    })?;
280                    dest.write_all(b"\n").map_err(|err| {
281                        IoError::new_internal(
282                            err,
283                            "Could not write linebreak after PipelineData to dest",
284                        )
285                    })?;
286                }
287                dest.flush().map_err(|err| {
288                    IoError::new_internal(err, "Could not flush PipelineData to dest")
289                })?;
290                Ok(())
291            }
292            PipelineData::ByteStream(stream, ..) => stream.write_to(dest),
293        }
294    }
295
296    /// Drain this [`PipelineData`] according to the current stdout [`OutDest`]s in `stack`.
297    ///
298    /// For [`OutDest::Pipe`] and [`OutDest::PipeSeparate`], this will return the [`PipelineData`]
299    /// as is. For [`OutDest::Value`], this will collect into a value and return it. For
300    /// [`OutDest::Print`], the [`PipelineData`] is drained and printed. Otherwise, the
301    /// [`PipelineData`] is drained, but only printed if it is the output of an external command.
302    pub fn drain_to_out_dests(
303        mut self,
304        engine_state: &EngineState,
305        stack: &mut Stack,
306    ) -> Result<Self, ShellError> {
307        match stack.pipe_stdout().unwrap_or(&OutDest::Inherit) {
308            OutDest::Print => {
309                self.print_table(engine_state, stack, false, false)?;
310                Ok(Self::Empty)
311            }
312            OutDest::Pipe | OutDest::PipeSeparate => Ok(self),
313            OutDest::Value => {
314                let metadata = self.take_metadata();
315                let span = self.span().unwrap_or(Span::unknown());
316                self.into_value(span).map(|val| Self::Value(val, metadata))
317            }
318            OutDest::File(file) => {
319                self.write_to(file.as_ref())?;
320                Ok(Self::Empty)
321            }
322            OutDest::Null | OutDest::Inherit => {
323                self.drain()?;
324                Ok(Self::Empty)
325            }
326        }
327    }
328
329    pub fn drain(self) -> Result<(), ShellError> {
330        match self {
331            Self::Empty => Ok(()),
332            Self::Value(Value::Error { error, .. }, ..) => Err(*error),
333            Self::Value(..) => Ok(()),
334            Self::ListStream(stream, ..) => stream.drain(),
335            Self::ByteStream(stream, ..) => stream.drain(),
336        }
337    }
338
339    /// Try convert from self into iterator
340    ///
341    /// It returns Err if the `self` cannot be converted to an iterator.
342    ///
343    /// The `span` should be the span of the command or operation that would raise an error.
344    pub fn into_iter_strict(self, span: Span) -> Result<PipelineIterator, ShellError> {
345        Ok(PipelineIterator(match self {
346            PipelineData::Value(value, ..) => {
347                let val_span = value.span();
348                match value {
349                    Value::List { vals, .. } => PipelineIteratorInner::ListStream(
350                        ListStream::new(vals.into_iter(), val_span, Signals::empty()).into_iter(),
351                    ),
352                    Value::Binary { val, .. } => PipelineIteratorInner::ListStream(
353                        ListStream::new(
354                            val.into_owned()
355                                .into_iter()
356                                .map(move |x| Value::int(x as i64, val_span)),
357                            val_span,
358                            Signals::empty(),
359                        )
360                        .into_iter(),
361                    ),
362                    Value::Range { val, .. } => PipelineIteratorInner::ListStream(
363                        ListStream::new(
364                            val.into_range_iter(val_span, Signals::empty()),
365                            val_span,
366                            Signals::empty(),
367                        )
368                        .into_iter(),
369                    ),
370                    // Handle iterable custom values by converting to base value first
371                    #[expect(deprecated)]
372                    Value::Custom { ref val, .. } if val.is_iterable() => {
373                        match val.to_base_value(val_span) {
374                            Ok(Value::List { vals, .. }) => PipelineIteratorInner::ListStream(
375                                ListStream::new(vals.into_iter(), val_span, Signals::empty())
376                                    .into_iter(),
377                            ),
378                            Ok(other) => {
379                                return Err(ShellError::OnlySupportsThisInputType {
380                                    exp_input_type: "list, binary, range, or byte stream".into(),
381                                    wrong_type: other.get_type().to_string(),
382                                    dst_span: span,
383                                    src_span: val_span,
384                                });
385                            }
386                            Err(err) => return Err(err),
387                        }
388                    }
389                    // Propagate errors by explicitly matching them before the final case.
390                    Value::Error { error, .. } => return Err(*error),
391                    other => {
392                        return Err(ShellError::OnlySupportsThisInputType {
393                            exp_input_type: "list, binary, range, or byte stream".into(),
394                            wrong_type: other.get_type().to_string(),
395                            dst_span: span,
396                            src_span: val_span,
397                        });
398                    }
399                }
400            }
401            PipelineData::ListStream(stream, ..) => {
402                PipelineIteratorInner::ListStream(stream.into_iter())
403            }
404            PipelineData::Empty => {
405                return Err(ShellError::OnlySupportsThisInputType {
406                    exp_input_type: "list, binary, range, or byte stream".into(),
407                    wrong_type: "null".into(),
408                    dst_span: span,
409                    src_span: span,
410                });
411            }
412            PipelineData::ByteStream(stream, ..) => {
413                if let Some(chunks) = stream.chunks() {
414                    PipelineIteratorInner::ByteStream(chunks)
415                } else {
416                    PipelineIteratorInner::Empty
417                }
418            }
419        }))
420    }
421
422    pub fn collect_string(self, separator: &str, config: &Config) -> Result<String, ShellError> {
423        match self {
424            PipelineData::Empty => Ok(String::new()),
425            PipelineData::Value(value, ..) => Ok(value.to_expanded_string(separator, config)),
426            PipelineData::ListStream(stream, ..) => Ok(stream.into_string(separator, config)),
427            PipelineData::ByteStream(stream, ..) => stream.into_string(),
428        }
429    }
430
431    /// Retrieves string from pipeline data.
432    ///
433    /// As opposed to `collect_string` this raises error rather than converting non-string values.
434    /// The `span` will be used if `ListStream` is encountered since it doesn't carry a span.
435    pub fn collect_string_strict(
436        self,
437        span: Span,
438    ) -> Result<(String, Span, Option<PipelineMetadata>), ShellError> {
439        match self {
440            PipelineData::Empty => Ok((String::new(), span, None)),
441            PipelineData::Value(Value::String { val, .. }, metadata) => Ok((val, span, metadata)),
442            PipelineData::Value(val, ..) => Err(ShellError::TypeMismatch {
443                err_message: "string".into(),
444                span: val.span(),
445            }),
446            PipelineData::ListStream(..) => Err(ShellError::TypeMismatch {
447                err_message: "string".into(),
448                span,
449            }),
450            PipelineData::ByteStream(stream, metadata) => {
451                let span = stream.span();
452                Ok((stream.into_string()?, span, metadata))
453            }
454        }
455    }
456
457    pub fn follow_cell_path(
458        self,
459        cell_path: &[PathMember],
460        head: Span,
461    ) -> Result<Value, ShellError> {
462        match self {
463            // FIXME: there are probably better ways of doing this
464            PipelineData::ListStream(stream, ..) => Value::list(stream.into_iter().collect(), head)
465                .follow_cell_path(cell_path)
466                .map(Cow::into_owned),
467            PipelineData::Value(v, ..) => v.follow_cell_path(cell_path).map(Cow::into_owned),
468            PipelineData::Empty => Err(ShellError::IncompatiblePathAccess {
469                type_name: "empty pipeline".to_string(),
470                span: head,
471            }),
472            PipelineData::ByteStream(stream, ..) => Err(ShellError::IncompatiblePathAccess {
473                type_name: stream.type_().describe().to_owned(),
474                span: stream.span(),
475            }),
476        }
477    }
478
479    /// Simplified mapper to help with simple values also. For full iterator support use `.into_iter()` instead
480    pub fn map<F>(self, mut f: F, signals: &Signals) -> Result<PipelineData, ShellError>
481    where
482        Self: Sized,
483        F: FnMut(Value) -> Value + 'static + Send,
484    {
485        match self {
486            PipelineData::Value(value, metadata) => {
487                let span = value.span();
488                let pipeline = match value {
489                    Value::List { vals, .. } => vals
490                        .into_iter()
491                        .map(f)
492                        .into_pipeline_data(span, signals.clone()),
493                    Value::Range { val, .. } => val
494                        .into_range_iter(span, Signals::empty())
495                        .map(f)
496                        .into_pipeline_data(span, signals.clone()),
497                    #[expect(deprecated)]
498                    Value::Custom { ref val, .. } if val.is_iterable() => {
499                        match val.to_base_value(span)? {
500                            Value::List { vals, .. } => vals
501                                .into_iter()
502                                .map(f)
503                                .into_pipeline_data(span, signals.clone()),
504                            Value::Range { val, .. } => val
505                                .into_range_iter(span, Signals::empty())
506                                .map(f)
507                                .into_pipeline_data(span, signals.clone()),
508                            value => match f(value) {
509                                Value::Error { error, .. } => return Err(*error),
510                                v => v.into_pipeline_data(),
511                            },
512                        }
513                    }
514                    value => match f(value) {
515                        Value::Error { error, .. } => return Err(*error),
516                        v => v.into_pipeline_data(),
517                    },
518                };
519                Ok(pipeline.set_metadata(metadata))
520            }
521            PipelineData::Empty => Ok(PipelineData::empty()),
522            PipelineData::ListStream(stream, metadata) => {
523                Ok(PipelineData::list_stream(stream.map(f), metadata))
524            }
525            PipelineData::ByteStream(stream, metadata) => {
526                Ok(f(stream.into_value()?).into_pipeline_data_with_metadata(metadata))
527            }
528        }
529    }
530
531    /// Simplified flatmapper. For full iterator support use `.into_iter()` instead
532    pub fn flat_map<U, F>(self, mut f: F, signals: &Signals) -> Result<PipelineData, ShellError>
533    where
534        Self: Sized,
535        U: IntoIterator<Item = Value> + 'static,
536        <U as IntoIterator>::IntoIter: 'static + Send,
537        F: FnMut(Value) -> U + 'static + Send,
538    {
539        match self {
540            PipelineData::Empty => Ok(PipelineData::empty()),
541            PipelineData::Value(value, metadata) => {
542                let span = value.span();
543                let pipeline = match value {
544                    Value::List { vals, .. } => vals
545                        .into_iter()
546                        .flat_map(f)
547                        .into_pipeline_data(span, signals.clone()),
548                    Value::Range { val, .. } => val
549                        .into_range_iter(span, Signals::empty())
550                        .flat_map(f)
551                        .into_pipeline_data(span, signals.clone()),
552                    #[expect(deprecated)]
553                    Value::Custom { ref val, .. } if val.is_iterable() => {
554                        match val.to_base_value(span)? {
555                            Value::List { vals, .. } => vals
556                                .into_iter()
557                                .flat_map(f)
558                                .into_pipeline_data(span, signals.clone()),
559                            Value::Range { val, .. } => val
560                                .into_range_iter(span, Signals::empty())
561                                .flat_map(f)
562                                .into_pipeline_data(span, signals.clone()),
563                            value => f(value)
564                                .into_iter()
565                                .into_pipeline_data(span, signals.clone()),
566                        }
567                    }
568                    value => f(value)
569                        .into_iter()
570                        .into_pipeline_data(span, signals.clone()),
571                };
572                Ok(pipeline.set_metadata(metadata))
573            }
574            PipelineData::ListStream(stream, metadata) => Ok(PipelineData::list_stream(
575                stream.modify(|iter| iter.flat_map(f)),
576                metadata,
577            )),
578            PipelineData::ByteStream(stream, metadata) => {
579                // TODO: is this behavior desired / correct ?
580                let span = stream.span();
581                let iter = match String::from_utf8(stream.into_bytes()?) {
582                    Ok(mut str) => {
583                        str.truncate(str.trim_end_matches(LINE_ENDING_PATTERN).len());
584                        f(Value::string(str, span))
585                    }
586                    Err(err) => f(Value::binary(err.into_bytes(), span)),
587                };
588                Ok(iter.into_iter().into_pipeline_data_with_metadata(
589                    span,
590                    signals.clone(),
591                    metadata,
592                ))
593            }
594        }
595    }
596
597    pub fn filter<F>(self, mut f: F, signals: &Signals) -> Result<PipelineData, ShellError>
598    where
599        Self: Sized,
600        F: FnMut(&Value) -> bool + 'static + Send,
601    {
602        match self {
603            PipelineData::Empty => Ok(PipelineData::empty()),
604            PipelineData::Value(value, metadata) => {
605                let span = value.span();
606                let pipeline = match value {
607                    Value::List { vals, .. } => vals
608                        .into_iter()
609                        .filter(f)
610                        .into_pipeline_data(span, signals.clone()),
611                    Value::Range { val, .. } => val
612                        .into_range_iter(span, Signals::empty())
613                        .filter(f)
614                        .into_pipeline_data(span, signals.clone()),
615                    #[expect(deprecated)]
616                    Value::Custom { ref val, .. } if val.is_iterable() => {
617                        match val.to_base_value(span)? {
618                            Value::List { vals, .. } => vals
619                                .into_iter()
620                                .filter(f)
621                                .into_pipeline_data(span, signals.clone()),
622                            Value::Range { val, .. } => val
623                                .into_range_iter(span, Signals::empty())
624                                .filter(f)
625                                .into_pipeline_data(span, signals.clone()),
626                            value => {
627                                if f(&value) {
628                                    value.into_pipeline_data()
629                                } else {
630                                    Value::nothing(span).into_pipeline_data()
631                                }
632                            }
633                        }
634                    }
635                    value => {
636                        if f(&value) {
637                            value.into_pipeline_data()
638                        } else {
639                            Value::nothing(span).into_pipeline_data()
640                        }
641                    }
642                };
643                Ok(pipeline.set_metadata(metadata))
644            }
645            PipelineData::ListStream(stream, metadata) => Ok(PipelineData::list_stream(
646                stream.modify(|iter| iter.filter(f)),
647                metadata,
648            )),
649            PipelineData::ByteStream(stream, metadata) => {
650                // TODO: is this behavior desired / correct ?
651                let span = stream.span();
652                let value = match String::from_utf8(stream.into_bytes()?) {
653                    Ok(mut str) => {
654                        str.truncate(str.trim_end_matches(LINE_ENDING_PATTERN).len());
655                        Value::string(str, span)
656                    }
657                    Err(err) => Value::binary(err.into_bytes(), span),
658                };
659                let value = if f(&value) {
660                    value
661                } else {
662                    Value::nothing(span)
663                };
664                Ok(value.into_pipeline_data_with_metadata(metadata))
665            }
666        }
667    }
668
669    /// Try to convert Value from Value::Range to Value::List.
670    /// This is useful to expand Value::Range into array notation, specifically when
671    /// converting `to json` or `to nuon`.
672    /// `1..3 | to XX -> [1,2,3]`
673    pub fn try_expand_range(self) -> Result<PipelineData, ShellError> {
674        match self {
675            PipelineData::Value(v, metadata) => {
676                let span = v.span();
677                match v {
678                    Value::Range { val, .. } => {
679                        match *val {
680                            Range::IntRange(range) => {
681                                if range.is_unbounded() {
682                                    return Err(ShellError::Generic(
683                                        GenericError::new(
684                                            "Cannot create range",
685                                            "Unbounded ranges are not allowed when converting to this format",
686                                            span,
687                                        )
688                                        .with_help(
689                                            "Consider using ranges with valid start and end point.",
690                                        ),
691                                    ));
692                                }
693                            }
694                            Range::FloatRange(range) => {
695                                if range.is_unbounded() {
696                                    return Err(ShellError::Generic(
697                                        GenericError::new(
698                                            "Cannot create range",
699                                            "Unbounded ranges are not allowed when converting to this format",
700                                            span,
701                                        )
702                                        .with_help(
703                                            "Consider using ranges with valid start and end point.",
704                                        ),
705                                    ));
706                                }
707                            }
708                        }
709                        let range_values: Vec<Value> =
710                            val.into_range_iter(span, Signals::empty()).collect();
711                        Ok(PipelineData::value(Value::list(range_values, span), None))
712                    }
713                    x => Ok(PipelineData::value(x, metadata)),
714                }
715            }
716            _ => Ok(self),
717        }
718    }
719
720    /// Consume and print self data immediately, formatted using table command.
721    ///
722    /// This does not respect the display_output hook. If a value is being printed out by a command,
723    /// this function should be used. Otherwise, `nu_cli::util::print_pipeline` should be preferred.
724    ///
725    /// `no_newline` controls if we need to attach newline character to output.
726    /// `to_stderr` controls if data is output to stderr, when the value is false, the data is output to stdout.
727    pub fn print_table(
728        self,
729        engine_state: &EngineState,
730        stack: &mut Stack,
731        no_newline: bool,
732        to_stderr: bool,
733    ) -> Result<(), ShellError> {
734        match self {
735            // Print byte streams directly as long as they aren't binary.
736            PipelineData::ByteStream(stream, ..) if stream.type_() != ByteStreamType::Binary => {
737                stream.print(to_stderr)
738            }
739            _ => {
740                // If the table function is in the declarations, then we can use it
741                // to create the table value that will be printed in the terminal
742                if let Some(decl_id) = engine_state.table_decl_id {
743                    let command = engine_state.get_decl(decl_id);
744                    if command.block_id().is_some() {
745                        self.write_all_and_flush(engine_state, no_newline, to_stderr)
746                    } else {
747                        let call = Call::new(Span::new(0, 0));
748                        let table = command.run(engine_state, stack, &(&call).into(), self)?;
749                        table.write_all_and_flush(engine_state, no_newline, to_stderr)
750                    }
751                } else {
752                    self.write_all_and_flush(engine_state, no_newline, to_stderr)
753                }
754            }
755        }
756    }
757
758    /// Consume and print self data without any extra formatting.
759    ///
760    /// This does not use the `table` command to format data, and also prints binary values and
761    /// streams in their raw format without generating a hexdump first.
762    ///
763    /// `no_newline` controls if we need to attach newline character to output.
764    /// `to_stderr` controls if data is output to stderr, when the value is false, the data is output to stdout.
765    pub fn print_raw(
766        self,
767        engine_state: &EngineState,
768        no_newline: bool,
769        to_stderr: bool,
770    ) -> Result<(), ShellError> {
771        let span = self.span();
772        if let PipelineData::Value(Value::Binary { val: bytes, .. }, _) = self {
773            if to_stderr {
774                write_all_and_flush(
775                    bytes.as_slice(),
776                    &mut std::io::stderr().lock(),
777                    "stderr",
778                    span,
779                    engine_state.signals(),
780                )?;
781            } else {
782                write_all_and_flush(
783                    bytes.as_slice(),
784                    &mut std::io::stdout().lock(),
785                    "stdout",
786                    span,
787                    engine_state.signals(),
788                )?;
789            }
790            Ok(())
791        } else {
792            self.write_all_and_flush(engine_state, no_newline, to_stderr)
793        }
794    }
795
796    fn write_all_and_flush(
797        self,
798        engine_state: &EngineState,
799        no_newline: bool,
800        to_stderr: bool,
801    ) -> Result<(), ShellError> {
802        let span = self.span();
803        if let PipelineData::ByteStream(stream, ..) = self {
804            // Copy ByteStreams directly
805            stream.print(to_stderr)
806        } else {
807            let config = engine_state.get_config();
808            for item in self {
809                let mut out = if let Value::Error { error, .. } = item {
810                    return Err(*error);
811                } else {
812                    item.to_expanded_string("\n", config)
813                };
814
815                if !no_newline {
816                    out.push('\n');
817                }
818
819                if to_stderr {
820                    write_all_and_flush(
821                        out,
822                        &mut std::io::stderr().lock(),
823                        "stderr",
824                        span,
825                        engine_state.signals(),
826                    )?;
827                } else {
828                    write_all_and_flush(
829                        out,
830                        &mut std::io::stdout().lock(),
831                        "stdout",
832                        span,
833                        engine_state.signals(),
834                    )?;
835                }
836            }
837
838            Ok(())
839        }
840    }
841
842    pub fn unsupported_input_error(
843        self,
844        expected_type: impl Into<String>,
845        span: Span,
846    ) -> ShellError {
847        match self {
848            PipelineData::Empty => ShellError::PipelineEmpty { dst_span: span },
849            PipelineData::Value(value, ..) => ShellError::OnlySupportsThisInputType {
850                exp_input_type: expected_type.into(),
851                wrong_type: value.get_type().get_non_specified_string(),
852                dst_span: span,
853                src_span: value.span(),
854            },
855            PipelineData::ListStream(stream, ..) => ShellError::OnlySupportsThisInputType {
856                exp_input_type: expected_type.into(),
857                wrong_type: "list (stream)".into(),
858                dst_span: span,
859                src_span: stream.span(),
860            },
861            PipelineData::ByteStream(stream, ..) => ShellError::OnlySupportsThisInputType {
862                exp_input_type: expected_type.into(),
863                wrong_type: stream.type_().describe().into(),
864                dst_span: span,
865                src_span: stream.span(),
866            },
867        }
868    }
869
870    // PipelineData might connect to a running process which has an exit status future
871    // Use this method to retrieve that future, it's useful for implementing `pipefail` feature.
872    #[cfg(feature = "os")]
873    pub fn clone_exit_status_future(&self) -> Option<ExitStatusGuard> {
874        match self {
875            PipelineData::Empty | PipelineData::Value(..) | PipelineData::ListStream(..) => None,
876            PipelineData::ByteStream(stream, ..) => match stream.source() {
877                ByteStreamSource::Read(..) | ByteStreamSource::File(..) => None,
878                ByteStreamSource::Child(c) => {
879                    let exit_future = c.clone_exit_status_future();
880                    let ignore_error = c.clone_ignore_error();
881                    Some(ExitStatusGuard::new(exit_future, ignore_error))
882                }
883            },
884        }
885    }
886}
887
888impl CompareTypes<Type> for PipelineData {
889    fn compare_types(&self, other: &Type) -> Option<TypeRelation> {
890        let self_ty = match self {
891            PipelineData::Empty => Type::Nothing,
892            PipelineData::ListStream(_, _) => Type::list(Type::Any),
893            PipelineData::ByteStream(stream, _) => stream.type_().into(),
894            PipelineData::Value(value, _) => return value.compare_types(other),
895        };
896        self_ty.compare_types(other)
897    }
898
899    /// Determine if the `PipelineData` can be assigned to `other`.
900    ///
901    /// This check makes no effort to collect a stream, so it may be a different result
902    /// than would be returned by calling [`Value::is_subtype_of()`] on the result of
903    /// [`.into_value()`](Self::into_value).
904    ///
905    /// A `ListStream` acts the same as an empty list type: it is a subtype of any [`list`](Type::List)
906    /// or [`table`](Type::Table) type. After converting to a value, it may become a more specific type.
907    /// For example, a `ListStream` is a subtype of `list<int>` and `list<string>`.
908    /// If calling [`.into_value()`](Self::into_value) results in a `list<int>`,
909    /// then the value would not be a subtype of `list<string>`, in contrast to the original `ListStream`.
910    ///
911    /// A `ByteStream` is a subtype of [`string`](Type::String) if it is coercible into a string.
912    /// Likewise, a `ByteStream` is a subtype of [`binary`](Type::Binary) if it is coercible into a binary value.
913    fn is_assignable_to(&self, dst: &Type) -> bool {
914        let self_ty = match self {
915            PipelineData::Empty => Type::Nothing,
916            PipelineData::ListStream(_, _) => Type::list(Type::Any),
917            PipelineData::ByteStream(stream, _) => stream.type_().into(),
918            PipelineData::Value(value, _) => return value.is_assignable_to(dst),
919        };
920        self_ty.is_assignable_to(dst)
921    }
922}
923
924pub fn write_all_and_flush<T>(
925    data: T,
926    destination: &mut impl Write,
927    destination_name: &str,
928    span: Option<Span>,
929    signals: &Signals,
930) -> Result<(), ShellError>
931where
932    T: AsRef<[u8]>,
933{
934    let io_error_map = |err: std::io::Error, location: &Location<'_>| {
935        let context = format!("Writing to {destination_name} failed");
936        match span {
937            None => IoError::new_internal_with_location(err, context, location),
938            Some(span) if span == Span::unknown() => {
939                IoError::new_internal_with_location(err, context, location)
940            }
941            Some(span) => IoError::new_with_additional_context(err, span, None, context),
942        }
943    };
944
945    let span = span.unwrap_or(Span::unknown());
946    const OUTPUT_CHUNK_SIZE: usize = 8192;
947    for chunk in data.as_ref().chunks(OUTPUT_CHUNK_SIZE) {
948        signals.check(&span)?;
949        destination
950            .write_all(chunk)
951            .map_err(|err| io_error_map(err, Location::caller()))?;
952    }
953    destination
954        .flush()
955        .map_err(|err| io_error_map(err, Location::caller()))?;
956    Ok(())
957}
958
959enum PipelineIteratorInner {
960    Empty,
961    Value(Value),
962    ListStream(crate::list_stream::IntoIter),
963    ByteStream(crate::byte_stream::Chunks),
964}
965
966pub struct PipelineIterator(PipelineIteratorInner);
967
968impl IntoIterator for PipelineData {
969    type Item = Value;
970
971    type IntoIter = PipelineIterator;
972
973    fn into_iter(self) -> Self::IntoIter {
974        PipelineIterator(match self {
975            PipelineData::Empty => PipelineIteratorInner::Empty,
976            PipelineData::Value(value, ..) => {
977                let span = value.span();
978                match value {
979                    Value::List { vals, signals, .. } => PipelineIteratorInner::ListStream(
980                        ListStream::new(
981                            vals.into_iter(),
982                            span,
983                            signals.unwrap_or_else(Signals::empty),
984                        )
985                        .into_iter(),
986                    ),
987                    Value::Range { val, signals, .. } => PipelineIteratorInner::ListStream(
988                        ListStream::new(
989                            val.into_range_iter(span, signals.unwrap_or_else(Signals::empty)),
990                            span,
991                            Signals::empty(),
992                        )
993                        .into_iter(),
994                    ),
995                    // Handle iterable custom values by converting to base value first
996                    #[expect(deprecated)]
997                    Value::Custom { ref val, .. } if val.is_iterable() => {
998                        match val.to_base_value(span) {
999                            Ok(Value::List { vals, signals, .. }) => {
1000                                PipelineIteratorInner::ListStream(
1001                                    ListStream::new(
1002                                        vals.into_iter(),
1003                                        span,
1004                                        signals.unwrap_or_else(Signals::empty),
1005                                    )
1006                                    .into_iter(),
1007                                )
1008                            }
1009                            Ok(other) => PipelineIteratorInner::Value(other),
1010                            Err(err) => PipelineIteratorInner::Value(Value::error(err, span)),
1011                        }
1012                    }
1013                    x => PipelineIteratorInner::Value(x),
1014                }
1015            }
1016            PipelineData::ListStream(stream, ..) => {
1017                PipelineIteratorInner::ListStream(stream.into_iter())
1018            }
1019            PipelineData::ByteStream(stream, ..) => stream.chunks().map_or(
1020                PipelineIteratorInner::Empty,
1021                PipelineIteratorInner::ByteStream,
1022            ),
1023        })
1024    }
1025}
1026
1027impl Iterator for PipelineIterator {
1028    type Item = Value;
1029
1030    fn next(&mut self) -> Option<Self::Item> {
1031        match &mut self.0 {
1032            PipelineIteratorInner::Empty => None,
1033            PipelineIteratorInner::Value(Value::Nothing { .. }, ..) => None,
1034            PipelineIteratorInner::Value(v, ..) => Some(std::mem::take(v)),
1035            PipelineIteratorInner::ListStream(stream, ..) => stream.next(),
1036            PipelineIteratorInner::ByteStream(stream) => stream.next().map(|x| match x {
1037                Ok(x) => x,
1038                Err(err) => Value::error(
1039                    err,
1040                    Span::unknown(), //FIXME: unclear where this span should come from
1041                ),
1042            }),
1043        }
1044    }
1045}
1046
1047pub trait IntoPipelineData {
1048    fn into_pipeline_data(self) -> PipelineData;
1049
1050    fn into_pipeline_data_with_metadata(
1051        self,
1052        metadata: impl Into<Option<PipelineMetadata>>,
1053    ) -> PipelineData;
1054}
1055
1056impl<V> IntoPipelineData for V
1057where
1058    V: Into<Value>,
1059{
1060    fn into_pipeline_data(self) -> PipelineData {
1061        PipelineData::value(self.into(), None)
1062    }
1063
1064    fn into_pipeline_data_with_metadata(
1065        self,
1066        metadata: impl Into<Option<PipelineMetadata>>,
1067    ) -> PipelineData {
1068        PipelineData::value(self.into(), metadata.into())
1069    }
1070}
1071
1072pub trait IntoInterruptiblePipelineData {
1073    fn into_pipeline_data(self, span: Span, signals: Signals) -> PipelineData;
1074    fn into_pipeline_data_with_metadata(
1075        self,
1076        span: Span,
1077        signals: Signals,
1078        metadata: impl Into<Option<PipelineMetadata>>,
1079    ) -> PipelineData;
1080}
1081
1082impl<I> IntoInterruptiblePipelineData for I
1083where
1084    I: IntoIterator + Send + 'static,
1085    I::IntoIter: Send + 'static,
1086    <I::IntoIter as Iterator>::Item: Into<Value>,
1087{
1088    fn into_pipeline_data(self, span: Span, signals: Signals) -> PipelineData {
1089        ListStream::new(self.into_iter().map(Into::into), span, signals).into()
1090    }
1091
1092    fn into_pipeline_data_with_metadata(
1093        self,
1094        span: Span,
1095        signals: Signals,
1096        metadata: impl Into<Option<PipelineMetadata>>,
1097    ) -> PipelineData {
1098        PipelineData::list_stream(
1099            ListStream::new(self.into_iter().map(Into::into), span, signals),
1100            metadata.into(),
1101        )
1102    }
1103}
1104
1105fn value_to_bytes(value: Value) -> Result<Vec<u8>, ShellError> {
1106    let bytes = match value {
1107        Value::String { val, .. } => val.into_bytes(),
1108        Value::Binary { val, .. } => val.into_owned(),
1109        Value::List { vals, .. } => {
1110            let val = vals
1111                .into_iter()
1112                .map(Value::coerce_into_string)
1113                .collect::<Result<Vec<String>, ShellError>>()?
1114                .join("\n")
1115                + "\n";
1116
1117            val.into_bytes()
1118        }
1119        // Propagate errors by explicitly matching them before the final case.
1120        Value::Error { error, .. } => return Err(*error),
1121        value => value.coerce_into_string()?.into_bytes(),
1122    };
1123    Ok(bytes)
1124}
1125
1126/// A wrapper to [`PipelineData`] which can also track exit status.
1127///
1128/// We use exit status tracking to implement the `pipefail` feature.
1129#[derive(Debug)]
1130pub struct PipelineExecutionData {
1131    pub body: PipelineData,
1132    #[cfg(feature = "os")]
1133    pub exit: Vec<Option<ExitStatusGuard>>,
1134    /// Whether this data was produced by an early `return` from the block, rather than by
1135    /// evaluating to the end of the block.
1136    ///
1137    /// The flag exists for a single consumer: top-level file evaluation reads it to detect a
1138    /// top-level `return` in a script and skip running `main`. Custom command calls and closure
1139    /// invocations instead clear it via
1140    /// [`eval_block_with_early_return`](https://docs.rs/nu-engine/latest/nu_engine/fn.eval_block_with_early_return.html),
1141    /// so it never leaks past a nested call and only ever reflects a `return` at the current level.
1142    pub early_return: bool,
1143}
1144
1145impl Deref for PipelineExecutionData {
1146    type Target = PipelineData;
1147
1148    fn deref(&self) -> &Self::Target {
1149        &self.body
1150    }
1151}
1152
1153impl DerefMut for PipelineExecutionData {
1154    fn deref_mut(&mut self) -> &mut Self::Target {
1155        &mut self.body
1156    }
1157}
1158
1159impl PipelineExecutionData {
1160    pub fn empty() -> Self {
1161        Self {
1162            body: PipelineData::empty(),
1163            #[cfg(feature = "os")]
1164            exit: vec![],
1165            early_return: false,
1166        }
1167    }
1168
1169    /// Mark this data as having been produced by an early `return`.
1170    pub fn with_early_return(mut self) -> Self {
1171        self.early_return = true;
1172        self
1173    }
1174}
1175
1176impl From<PipelineData> for PipelineExecutionData {
1177    #[cfg(feature = "os")]
1178    fn from(value: PipelineData) -> Self {
1179        let value_span = value.span().unwrap_or_else(Span::unknown);
1180        let exit_status_future = value
1181            .clone_exit_status_future()
1182            .map(|f| f.with_span(value_span));
1183        Self {
1184            body: value,
1185            exit: vec![exit_status_future],
1186            early_return: false,
1187        }
1188    }
1189
1190    #[cfg(not(feature = "os"))]
1191    fn from(value: PipelineData) -> Self {
1192        Self {
1193            body: value,
1194            early_return: false,
1195        }
1196    }
1197}
nu_protocol/pipeline/pipeline_data.rs

nu_protocol/pipeline/
pipeline_data.rs
