Merge pull request #23 from wycats/improved-streams

Improved streams

Cargo.lock

@@ -143,6 +143,15 @@ name = "byteorder"
 version = "1.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
+[[package]]
+name = "bytes"
+version = "0.4.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "byteorder 1.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "iovec 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "cc"
 version = "1.0.37"
@@ -285,6 +294,15 @@ dependencies = [
  "crossbeam-utils 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "crossbeam-deque"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossbeam-epoch 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossbeam-utils 0.6.5 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "crossbeam-epoch"
 version = "0.3.1"
@@ -299,6 +317,27 @@ dependencies = [
  "scopeguard 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "crossbeam-epoch"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "arrayvec 0.4.10 (registry+https://github.com/rust-lang/crates.io-index)",
+ "cfg-if 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossbeam-utils 0.6.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "lazy_static 1.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "memoffset 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "scopeguard 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "crossbeam-queue"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossbeam-utils 0.6.5 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "crossbeam-utils"
 version = "0.2.2"
@@ -656,11 +695,83 @@ version = "0.1.27"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
 [[package]]
-name = "futures-core"
+name = "futures-channel-preview"
-version = "0.2.1"
+version = "0.3.0-alpha.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
- "either 1.5.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-core-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "futures-core-preview"
+version = "0.3.0-alpha.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
+[[package]]
+name = "futures-executor-preview"
+version = "0.3.0-alpha.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "futures-channel-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-core-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-util-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "num_cpus 1.10.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "futures-io-preview"
+version = "0.3.0-alpha.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "futures-core-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "futures-preview"
+version = "0.3.0-alpha.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "futures-channel-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-core-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-executor-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-io-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-sink-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-util-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "futures-sink-preview"
+version = "0.3.0-alpha.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "futures-channel-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-core-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "futures-util-preview"
+version = "0.3.0-alpha.16"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "futures 0.1.27 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-channel-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-core-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-io-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-sink-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "memchr 2.2.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-io 0.1.12 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "futures_codec"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "bytes 0.4.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -686,6 +797,15 @@ name = "indexmap"
 version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
+[[package]]
+name = "iovec"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "libc 0.2.55 (registry+https://github.com/rust-lang/crates.io-index)",
+ "winapi 0.2.8 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "itertools"
 version = "0.8.0"
@@ -859,17 +979,21 @@ dependencies = [
  "cursive 0.12.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "derive-new 0.5.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "dunce 1.0.0 (registry+https://github.com/rust-lang/crates.io-index)",
- "futures 0.1.27 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
- "futures-core 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures-sink-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures_codec 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "indexmap 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "itertools 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "nom 5.0.0-beta1 (registry+https://github.com/rust-lang/crates.io-index)",
  "ordered-float 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "pancurses 0.16.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "prettyprint 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "prettytable-rs 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "rustyline 4.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "subprocess 0.1.18 (registry+https://github.com/rust-lang/crates.io-index)",
  "sysinfo 0.8.4 (registry+https://github.com/rust-lang/crates.io-index)",
+ "term 0.5.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-fs 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -1031,6 +1155,11 @@ dependencies = [
  "libc 0.2.55 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "pin-utils"
+version = "0.1.0-alpha.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
 [[package]]
 name = "pkg-config"
 version = "0.3.14"
@@ -1395,6 +1524,11 @@ dependencies = [
  "libc 0.2.55 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "slab"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+
 [[package]]
 name = "smallvec"
 version = "0.6.9"
@@ -1566,6 +1700,51 @@ dependencies = [
  "winapi 0.3.7 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
+[[package]]
+name = "tokio-executor"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossbeam-utils 0.6.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures 0.1.27 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "tokio-fs"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "futures 0.1.27 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-io 0.1.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-threadpool 0.1.14 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "tokio-io"
+version = "0.1.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "bytes 0.4.12 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures 0.1.27 (registry+https://github.com/rust-lang/crates.io-index)",
+ "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
+name = "tokio-threadpool"
+version = "0.1.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "crossbeam-deque 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossbeam-queue 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "crossbeam-utils 0.6.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "futures 0.1.27 (registry+https://github.com/rust-lang/crates.io-index)",
+ "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "num_cpus 1.10.0 (registry+https://github.com/rust-lang/crates.io-index)",
+ "rand 0.6.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-executor 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
 [[package]]
 name = "toml"
 version = "0.5.1"
@@ -1720,6 +1899,7 @@ dependencies = [
 "checksum build_const 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "39092a32794787acd8525ee150305ff051b0aa6cc2abaf193924f5ab05425f39"
 "checksum byte-unit 2.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "6754bb4703aa167bed5381f0c6842f1cc31a9ecde3b9443f726dde3ad3afb841"
 "checksum byteorder 1.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "a019b10a2a7cdeb292db131fc8113e57ea2a908f6e7894b0c3c671893b65dbeb"
+"checksum bytes 0.4.12 (registry+https://github.com/rust-lang/crates.io-index)" = "206fdffcfa2df7cbe15601ef46c813fce0965eb3286db6b56c583b814b51c81c"
 "checksum cc 1.0.37 (registry+https://github.com/rust-lang/crates.io-index)" = "39f75544d7bbaf57560d2168f28fd649ff9c76153874db88bdbdfd839b1a7e7d"
 "checksum cfg-if 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)" = "b486ce3ccf7ffd79fdeb678eac06a9e6c09fc88d33836340becb8fffe87c5e33"
 "checksum chrono 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)" = "45912881121cb26fad7c38c17ba7daa18764771836b34fab7d3fbd93ed633878"
@@ -1736,7 +1916,10 @@ dependencies = [
 "checksum crc32fast 1.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "ba125de2af0df55319f41944744ad91c71113bf74a4646efff39afe1f6842db1"
 "checksum crossbeam-channel 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)" = "0f0ed1a4de2235cabda8558ff5840bffb97fcb64c97827f354a451307df5f72b"
 "checksum crossbeam-deque 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "f739f8c5363aca78cfb059edf753d8f0d36908c348f3d8d1503f03d8b75d9cf3"
+"checksum crossbeam-deque 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b18cd2e169ad86297e6bc0ad9aa679aee9daa4f19e8163860faf7c164e4f5a71"
 "checksum crossbeam-epoch 0.3.1 (registry+https://github.com/rust-lang/crates.io-index)" = "927121f5407de9956180ff5e936fe3cf4324279280001cd56b669d28ee7e9150"
+"checksum crossbeam-epoch 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)" = "04c9e3102cc2d69cd681412141b390abd55a362afc1540965dad0ad4d34280b4"
+"checksum crossbeam-queue 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7c979cd6cfe72335896575c6b5688da489e420d36a27a0b9eb0c73db574b4a4b"
 "checksum crossbeam-utils 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "2760899e32a1d58d5abb31129f8fae5de75220bc2176e77ff7c627ae45c918d9"
 "checksum crossbeam-utils 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)" = "677d453a17e8bd2b913fa38e8b9cf04bcdbb5be790aa294f2389661d72036015"
 "checksum crossbeam-utils 0.6.5 (registry+https://github.com/rust-lang/crates.io-index)" = "f8306fcef4a7b563b76b7dd949ca48f52bc1141aa067d2ea09565f3e2652aa5c"
@@ -1776,11 +1959,19 @@ dependencies = [
 "checksum fnv 1.0.6 (registry+https://github.com/rust-lang/crates.io-index)" = "2fad85553e09a6f881f739c29f0b00b0f01357c743266d478b68951ce23285f3"
 "checksum fuchsia-cprng 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "a06f77d526c1a601b7c4cdd98f54b5eaabffc14d5f2f0296febdc7f357c6d3ba"
 "checksum futures 0.1.27 (registry+https://github.com/rust-lang/crates.io-index)" = "a2037ec1c6c1c4f79557762eab1f7eae1f64f6cb418ace90fae88f0942b60139"
-"checksum futures-core 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "7455c91eb2eae38f33b013f77ebe766c75761af333efd9d550e154045c63e225"
+"checksum futures-channel-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)" = "4cd523712fc272e9b714669165a2832debee5a5b7e409bfccdc7c0d5cd0cf07a"
+"checksum futures-core-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)" = "719770f328642b657b849856bb5a607db9538dd5bb3000122e5ead55d0a58c36"
+"checksum futures-executor-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)" = "315dc58c908535d059576a329b86cd185933433382cfcd394fb2fa353330de03"
+"checksum futures-io-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)" = "cca0bf7a1f39c9d32b797b0def93d5932aa71796236aad6b549bac6f7df159a3"
+"checksum futures-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)" = "fcfeac5f016a4b5835bb93eb7961f50a64f0e001207562703d9ddf4109d7b263"
+"checksum futures-sink-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)" = "49dcfdacd6b5974ca0b9b78bc38ffd1071da0206179735c3df82e279f5b784e4"
+"checksum futures-util-preview 0.3.0-alpha.16 (registry+https://github.com/rust-lang/crates.io-index)" = "f7a0451b9c5047c2b9ab93425ffd0793165511e93c04b977cd45fbd41c6e34b2"
+"checksum futures_codec 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "b60f48aa03e365df015d2fbf0b79f17b440350c268a5e20305da17b394adcc1e"
 "checksum hashbrown 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)" = "570178d5e4952010d138b0f1d581271ff3a02406d990f887d1e87e3d6e43b0ac"
 "checksum humantime 1.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "3ca7e5f2e110db35f93b837c81797f3714500b81d517bf20c431b16d3ca4f114"
 "checksum ident_case 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39"
 "checksum indexmap 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7e81a7c05f79578dbc15793d8b619db9ba32b4577003ef3af1a91c416798c58d"
+"checksum iovec 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "dbe6e417e7d0975db6512b90796e8ce223145ac4e33c377e4a42882a0e88bb08"
 "checksum itertools 0.8.0 (registry+https://github.com/rust-lang/crates.io-index)" = "5b8467d9c1cebe26feb08c640139247fac215782d35371ade9a2136ed6085358"
 "checksum itoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)" = "501266b7edd0174f8530248f87f99c88fbe60ca4ef3dd486835b8d8d53136f7f"
 "checksum kernel32-sys 0.2.2 (registry+https://github.com/rust-lang/crates.io-index)" = "7507624b29483431c0ba2d82aece8ca6cdba9382bff4ddd0f7490560c056098d"
@@ -1819,6 +2010,7 @@ dependencies = [
 "checksum parking_lot_core 0.5.0 (registry+https://github.com/rust-lang/crates.io-index)" = "cb88cb1cb3790baa6776844f968fea3be44956cf184fa1be5a03341f5491278c"
 "checksum parse-zoneinfo 0.2.0 (registry+https://github.com/rust-lang/crates.io-index)" = "089a398ccdcdd77b8c38909d5a1e4b67da1bc4c9dbfe6d5b536c828eddb779e5"
 "checksum pdcurses-sys 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)" = "084dd22796ff60f1225d4eb6329f33afaf4c85419d51d440ab6b8c6f4529166b"
+"checksum pin-utils 0.1.0-alpha.4 (registry+https://github.com/rust-lang/crates.io-index)" = "5894c618ce612a3fa23881b152b608bafb8c56cfc22f434a3ba3120b40f7b587"
 "checksum pkg-config 0.3.14 (registry+https://github.com/rust-lang/crates.io-index)" = "676e8eb2b1b4c9043511a9b7bea0915320d7e502b0a079fb03f9635a5252b18c"
 "checksum plist 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)" = "f4739851c08dd9a62a78beff2edf1a438517268b2c563c42fc6d9d3139e42d2a"
 "checksum prettyprint 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)" = "2705417f8aa07cb6308db42e55623479c1c9667942a4d5e4174c684e5da5590d"
@@ -1861,6 +2053,7 @@ dependencies = [
 "checksum shell-words 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "39acde55a154c4cd3ae048ac78cc21c25f3a0145e44111b523279113dce0d94a"
 "checksum signal-hook 0.1.9 (registry+https://github.com/rust-lang/crates.io-index)" = "72ab58f1fda436857e6337dcb6a5aaa34f16c5ddc87b3a8b6ef7a212f90b9c5a"
 "checksum signal-hook-registry 1.0.1 (registry+https://github.com/rust-lang/crates.io-index)" = "cded4ffa32146722ec54ab1f16320568465aa922aa9ab4708129599740da85d7"
+"checksum slab 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)" = "c111b5bd5695e56cffe5129854aa230b39c93a305372fdbb2668ca2394eea9f8"
 "checksum smallvec 0.6.9 (registry+https://github.com/rust-lang/crates.io-index)" = "c4488ae950c49d403731982257768f48fada354a5203fe81f9bb6f43ca9002be"
 "checksum stable_deref_trait 1.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "dba1a27d3efae4351c8051072d619e3ade2820635c3958d826bfea39d59b54c8"
 "checksum stackvector 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "c049c77bf85fbc036484c97b008276d539d9ebff9dfbde37b632ebcd5b8746b6"
@@ -1880,6 +2073,10 @@ dependencies = [
 "checksum textwrap 0.11.0 (registry+https://github.com/rust-lang/crates.io-index)" = "d326610f408c7a4eb6f51c37c330e496b08506c9457c9d34287ecc38809fb060"
 "checksum thread_local 0.3.6 (registry+https://github.com/rust-lang/crates.io-index)" = "c6b53e329000edc2b34dbe8545fd20e55a333362d0a321909685a19bd28c3f1b"
 "checksum time 0.1.42 (registry+https://github.com/rust-lang/crates.io-index)" = "db8dcfca086c1143c9270ac42a2bbd8a7ee477b78ac8e45b19abfb0cbede4b6f"
+"checksum tokio-executor 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)" = "83ea44c6c0773cc034771693711c35c677b4b5a4b21b9e7071704c54de7d555e"
+"checksum tokio-fs 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "3fe6dc22b08d6993916647d108a1a7d15b9cd29c4f4496c62b92c45b5041b7af"
+"checksum tokio-io 0.1.12 (registry+https://github.com/rust-lang/crates.io-index)" = "5090db468dad16e1a7a54c8c67280c5e4b544f3d3e018f0b913b400261f85926"
+"checksum tokio-threadpool 0.1.14 (registry+https://github.com/rust-lang/crates.io-index)" = "72558af20be886ea124595ea0f806dd5703b8958e4705429dd58b3d8231f72f2"
 "checksum toml 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b8c96d7873fa7ef8bdeb3a9cda3ac48389b4154f32b9803b4bc26220b677b039"
 "checksum ucd-util 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "535c204ee4d8434478593480b8f86ab45ec9aae0e83c568ca81abf0fd0e88f86"
 "checksum unicode-segmentation 1.3.0 (registry+https://github.com/rust-lang/crates.io-index)" = "1967f4cdfc355b37fd76d2a954fb2ed3871034eb4f26d60537d88795cfc332a9"

Cargo.toml

@@ -21,11 +21,18 @@ conch-parser = "0.1.1"
 nom = "5.0.0-beta1"
 subprocess = "0.1.18"
 dunce = "1.0.0"
-futures = "0.1.27"
-futures-core = "0.2.1"
 indexmap = "1.0.2"
 chrono-humanize = "0.0.11"
 byte-unit = "2.1.0"
 ordered-float = "1.0.2"
 prettyprint = "0.6.0"
 cursive = { version = "0.12.0", features = ["pancurses-backend"], default-features = false }
+futures-preview = { version = "0.3.0-alpha.16", features = ["compat", "io-compat"] }
+futures-sink-preview = "0.3.0-alpha.16"
+tokio-fs = "0.1.6"
+futures_codec = "0.2.2"
+term = "0.5.2"
+
+[dependencies.pancurses]
+version = "0.16"
+features = ["win32a"]

rustfmt.toml

@@ -0,0 +1 @@
+edition = "2018"

src/cli.rs

@@ -0,0 +1,357 @@
+use crate::prelude::*;
+
+use crate::commands::classified::{
+    ClassifiedCommand, ClassifiedInputStream, ExternalCommand, InternalCommand, StreamNext,
+};
+use crate::context::Context;
+crate use crate::errors::ShellError;
+crate use crate::format::{EntriesListView, GenericView};
+use crate::object::Value;
+use crate::stream::empty_stream;
+
+use rustyline::error::ReadlineError;
+use rustyline::{self, ColorMode, Config, Editor};
+use std::collections::VecDeque;
+use std::error::Error;
+use std::iter::Iterator;
+use std::sync::Arc;
+
+#[derive(Debug)]
+pub enum MaybeOwned<'a, T> {
+    Owned(T),
+    Borrowed(&'a T),
+}
+
+impl<T> MaybeOwned<'a, T> {
+    crate fn borrow(&self) -> &T {
+        match self {
+            MaybeOwned::Owned(v) => v,
+            MaybeOwned::Borrowed(v) => v,
+        }
+    }
+}
+
+pub async fn cli() -> Result<(), Box<Error>> {
+    let config = Config::builder().color_mode(ColorMode::Forced).build();
+    let h = crate::shell::Helper::new();
+    let mut rl: Editor<crate::shell::Helper> = Editor::with_config(config);
+
+    #[cfg(windows)]
+    {
+        let _ = ansi_term::enable_ansi_support();
+    }
+
+    rl.set_helper(Some(h));
+    if rl.load_history("history.txt").is_err() {
+        println!("No previous history.");
+    }
+
+    let mut context = Context::basic()?;
+
+    {
+        use crate::commands::*;
+
+        context.add_commands(vec![
+            ("format-list", Arc::new(format_list)),
+            ("ps", Arc::new(ps::ps)),
+            ("ls", Arc::new(ls::ls)),
+            ("cd", Arc::new(cd::cd)),
+            ("view", Arc::new(view::view)),
+            ("skip", Arc::new(skip::skip)),
+            ("first", Arc::new(take::take)),
+            ("select", Arc::new(select::select)),
+            ("reject", Arc::new(reject::reject)),
+            ("to-array", Arc::new(to_array::to_array)),
+            ("where", Arc::new(where_::r#where)),
+            ("sort-by", Arc::new(sort_by::sort_by)),
+        ]);
+    }
+
+    loop {
+        let readline = rl.readline(&format!(
+            "{}> ",
+            context.env.lock().unwrap().cwd().display().to_string()
+        ));
+
+        match process_line(readline, &mut context).await {
+            LineResult::Success(line) => {
+                rl.add_history_entry(line.clone());
+            }
+
+            LineResult::Error(err) => {
+                context.host.lock().unwrap().stdout(&err);
+            }
+
+            LineResult::Break => {
+                break;
+            }
+
+            LineResult::FatalError(err) => {
+                context
+                    .host
+                    .lock()
+                    .unwrap()
+                    .stdout(&format!("A surprising fatal error occurred.\n{:?}", err));
+            }
+        }
+    }
+    rl.save_history("history.txt").unwrap();
+
+    Ok(())
+}
+
+enum LineResult {
+    Success(String),
+    Error(String),
+    Break,
+
+    #[allow(unused)]
+    FatalError(ShellError),
+}
+
+impl std::ops::Try for LineResult {
+    type Ok = Option<String>;
+    type Error = ShellError;
+
+    fn into_result(self) -> Result<Option<String>, ShellError> {
+        match self {
+            LineResult::Success(s) => Ok(Some(s)),
+            LineResult::Error(s) => Err(ShellError::string(s)),
+            LineResult::Break => Ok(None),
+            LineResult::FatalError(err) => Err(err),
+        }
+    }
+    fn from_error(v: ShellError) -> Self {
+        LineResult::Error(v.to_string())
+    }
+
+    fn from_ok(v: Option<String>) -> Self {
+        match v {
+            None => LineResult::Break,
+            Some(v) => LineResult::Success(v),
+        }
+    }
+}
+
+async fn process_line(readline: Result<String, ReadlineError>, ctx: &mut Context) -> LineResult {
+    match &readline {
+        Ok(line) if line.trim() == "exit" => LineResult::Break,
+
+        Ok(line) if line.trim() == "" => LineResult::Success(line.clone()),
+
+        Ok(line) => {
+            let result = match crate::parser::shell_parser(&line) {
+                Err(err) => {
+                    return LineResult::Error(format!("{:?}", err));
+                }
+
+                Ok(val) => val,
+            };
+
+            let parsed: Result<Vec<_>, _> = result
+                .1
+                .into_iter()
+                .map(|item| classify_command(&item, ctx))
+                .collect();
+
+            let parsed = parsed?;
+
+            let mut input = ClassifiedInputStream::new();
+
+            let mut iter = parsed.into_iter().peekable();
+
+            loop {
+                let item: Option<ClassifiedCommand> = iter.next();
+                let next: Option<&ClassifiedCommand> = iter.peek();
+
+                input = match (item, next) {
+                    (None, _) => break,
+
+                    (
+                        Some(ClassifiedCommand::Internal(left)),
+                        Some(ClassifiedCommand::Internal(_)),
+                    ) => match left.run(ctx, input).await {
+                        Ok(val) => ClassifiedInputStream::from_input_stream(val),
+                        Err(err) => return LineResult::Error(format!("{}", err.description())),
+                    },
+
+                    (Some(ClassifiedCommand::Internal(left)), None) => {
+                        match left.run(ctx, input).await {
+                            Ok(val) => ClassifiedInputStream::from_input_stream(val),
+                            Err(err) => return LineResult::Error(format!("{}", err.description())),
+                        }
+                    }
+
+                    (
+                        Some(ClassifiedCommand::External(left)),
+                        Some(ClassifiedCommand::External(_)),
+                    ) => match left.run(ctx, input, StreamNext::External).await {
+                        Ok(val) => val,
+                        Err(err) => return LineResult::Error(format!("{}", err.description())),
+                    },
+
+                    (
+                        Some(ClassifiedCommand::Internal(_)),
+                        Some(ClassifiedCommand::External(_)),
+                    ) => unimplemented!(),
+
+                    (
+                        Some(ClassifiedCommand::External(left)),
+                        Some(ClassifiedCommand::Internal(_)),
+                    ) => match left.run(ctx, input, StreamNext::Internal).await {
+                        Ok(val) => val,
+                        Err(err) => return LineResult::Error(format!("{}", err.description())),
+                    },
+
+                    (Some(ClassifiedCommand::External(left)), None) => {
+                        match left.run(ctx, input, StreamNext::Last).await {
+                            Ok(val) => val,
+                            Err(err) => return LineResult::Error(format!("{}", err.description())),
+                        }
+                    }
+                }
+                // input = match process_command(item.clone(), input, ctx).await {
+                //     Ok(val) => val,
+                //     Err(err) => return LineResult::Error(format!("{}", err.description())),
+                // };
+            }
+
+            let input_vec: VecDeque<_> = input.objects.collect().await;
+
+            if input_vec.len() > 0 {
+                if equal_shapes(&input_vec) {
+                    let array = crate::commands::stream_to_array(input_vec.boxed()).await;
+                    let args = CommandArgs::from_context(ctx, vec![], array);
+                    let mut result = format(args);
+                    let mut vec = vec![];
+                    vec.send_all(&mut result).await?;
+                } else {
+                    let args = CommandArgs::from_context(ctx, vec![], input_vec.boxed());
+                    format(args).collect::<Vec<_>>().await;
+                }
+            }
+
+            LineResult::Success(line.to_string())
+        }
+        Err(ReadlineError::Interrupted) => {
+            println!("CTRL-C");
+            LineResult::Break
+        }
+        Err(ReadlineError::Eof) => {
+            println!("CTRL-D");
+            LineResult::Break
+        }
+        Err(err) => {
+            println!("Error: {:?}", err);
+            LineResult::Break
+        }
+    }
+}
+
+// async fn process_command(
+//     parsed: Vec<crate::parser::Item>,
+//     input: InputStream,
+//     context: &mut Context,
+// ) -> Result<InputStream, ShellError> {
+//     let command = classify_command(&parsed, context)?;
+
+//     command.run(context, input).await
+// }
+
+fn classify_command(
+    command: &[crate::parser::Item],
+    context: &Context,
+) -> Result<ClassifiedCommand, ShellError> {
+    let command_name = &command[0].name()?;
+
+    let arg_list: Vec<Value> = command[1..].iter().map(|i| i.as_value()).collect();
+    let arg_list_strings: Vec<String> = command[1..].iter().map(|i| i.print()).collect();
+
+    match *command_name {
+        other => match context.has_command(*command_name) {
+            true => {
+                let command = context.get_command(command_name);
+                Ok(ClassifiedCommand::Internal(InternalCommand {
+                    command,
+                    args: arg_list,
+                }))
+            }
+            false => Ok(ClassifiedCommand::External(ExternalCommand {
+                name: other.to_string(),
+                args: arg_list_strings,
+            })),
+        },
+    }
+}
+
+crate fn format(args: CommandArgs) -> OutputStream {
+    let host = args.host.clone();
+    let input = args.input.map(|a| a.copy());
+    let input = input.collect::<Vec<_>>();
+
+    input
+        .then(move |input| {
+            let last = input.len() - 1;
+            let mut host = host.lock().unwrap();
+            for (i, item) in input.iter().enumerate() {
+                let view = GenericView::new(item);
+                crate::format::print_view(&view, &mut *host);
+
+                if last != i {
+                    println!("");
+                }
+            }
+
+            futures::future::ready(empty_stream())
+        })
+        .flatten_stream()
+        .boxed()
+}
+
+crate fn format_list(args: CommandArgs) -> Result<OutputStream, ShellError> {
+    let host = args.host.clone();
+    // let input = args.input.map(|a| a.copy());
+    // let input = input.collect::<Vec<_>>();
+
+    let view = EntriesListView::from_stream(args.input);
+
+    Ok(view
+        .then(move |view| {
+            crate::format::print_view(&view, &mut *host.lock().unwrap());
+
+            futures::future::ready(empty_stream())
+        })
+        .flatten_stream()
+        .boxed())
+
+    // Ok(empty_stream())
+    // Ok(args
+    //     .input
+    //     .map(|input| {
+    //         let view = EntriesListView::from_stream(input);
+    //         crate::format::print_view(&view, &mut *args.host.lock().unwrap());
+    //     })
+    //     .collect()
+    //     .then(|_| empty_stream())
+    //     .flatten_stream()
+    //     .boxed())
+}
+
+fn equal_shapes(input: &VecDeque<Value>) -> bool {
+    let mut items = input.iter();
+
+    let item = match items.next() {
+        Some(item) => item,
+        None => return false,
+    };
+
+    let desc = item.data_descriptors();
+
+    for item in items {
+        if desc != item.data_descriptors() {
+            return false;
+        }
+    }
+
+    true
+}

src/commands/cd.rs

@@ -2,18 +2,18 @@ use crate::errors::ShellError;
 use crate::prelude::*;
 use std::env;
 
-pub fn cd(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
+pub fn cd(args: CommandArgs) -> Result<OutputStream, ShellError> {
     let target = match args.args.first() {
         // TODO: This needs better infra
         None => return Err(ShellError::string(format!("cd must take one arg"))),
         Some(v) => v.as_string()?.clone(),
     };
 
-    let cwd = args.env.cwd().to_path_buf();
+    let cwd = args.env.lock().unwrap().cwd().to_path_buf();
 
     let mut stream = VecDeque::new();
     let path = dunce::canonicalize(cwd.join(&target).as_path())?;
     let _ = env::set_current_dir(&path);
     stream.push_back(ReturnValue::change_cwd(path));
-    Ok(stream)
+    Ok(stream.boxed())
 }

src/commands/classified.rs

@@ -1,44 +1,75 @@
 use crate::prelude::*;
+use futures::compat::AsyncRead01CompatExt;
+use futures_codec::{Framed, LinesCodec};
 use std::sync::Arc;
 use subprocess::Exec;
 
+crate struct ClassifiedInputStream {
+    crate objects: InputStream,
+    crate stdin: Option<std::fs::File>,
+}
+
+impl ClassifiedInputStream {
+    crate fn new() -> ClassifiedInputStream {
+        ClassifiedInputStream {
+            objects: VecDeque::new().boxed(),
+            stdin: None,
+        }
+    }
+
+    crate fn from_input_stream(stream: InputStream) -> ClassifiedInputStream {
+        ClassifiedInputStream {
+            objects: stream,
+            stdin: None,
+        }
+    }
+
+    crate fn from_stdout(stdout: std::fs::File) -> ClassifiedInputStream {
+        ClassifiedInputStream {
+            objects: VecDeque::new().boxed(),
+            stdin: Some(stdout),
+        }
+    }
+}
+
 crate enum ClassifiedCommand {
     Internal(InternalCommand),
     External(ExternalCommand),
 }
 
 impl ClassifiedCommand {
-    crate fn run(
+    crate async fn run(
         self,
-        input: VecDeque<Value>,
+        context: &mut Context,
-        context: &mut crate::Context,
+        input: ClassifiedInputStream,
-    ) -> Result<VecDeque<Value>, ShellError> {
+    ) -> Result<InputStream, ShellError> {
         match self {
             ClassifiedCommand::Internal(internal) => {
-                let result = context.run_command(internal.command, internal.args, input)?;
+                let result = context.run_command(internal.command, internal.args, input.objects)?;
+                let env = context.env.clone();
 
-                let mut next = VecDeque::new();
+                let stream = result.filter_map(move |v| match v {
-
-                for v in result {
-                    match v {
                     ReturnValue::Action(action) => match action {
-                            crate::CommandAction::ChangeCwd(cwd) => context.env.cwd = cwd,
+                        CommandAction::ChangeCwd(cwd) => {
+                            env.lock().unwrap().cwd = cwd;
+                            futures::future::ready(None)
+                        }
                     },
 
-                        ReturnValue::Value(v) => next.push_back(v),
+                    ReturnValue::Value(v) => futures::future::ready(Some(v)),
-                    }
+                });
-                }
 
-                Ok(next)
+                Ok(stream.boxed() as InputStream)
             }
 
             ClassifiedCommand::External(external) => {
                 Exec::shell(&external.name)
                     .args(&external.args)
-                    .cwd(context.env.cwd())
+                    .cwd(context.env.lock().unwrap().cwd())
                     .join()
                     .unwrap();
-                Ok(VecDeque::new())
+
+                Ok(VecDeque::new().boxed() as InputStream)
             }
         }
     }
@@ -49,7 +80,90 @@ crate struct InternalCommand {
     crate args: Vec<Value>,
 }
 
+impl InternalCommand {
+    crate async fn run(
+        self,
+        context: &mut Context,
+        input: ClassifiedInputStream,
+    ) -> Result<InputStream, ShellError> {
+        let result = context.run_command(self.command, self.args, input.objects)?;
+        let env = context.env.clone();
+
+        let stream = result.filter_map(move |v| match v {
+            ReturnValue::Action(action) => match action {
+                CommandAction::ChangeCwd(cwd) => {
+                    env.lock().unwrap().cwd = cwd;
+                    futures::future::ready(None)
+                }
+            },
+
+            ReturnValue::Value(v) => futures::future::ready(Some(v)),
+        });
+
+        Ok(stream.boxed() as InputStream)
+    }
+}
+
 crate struct ExternalCommand {
     crate name: String,
     crate args: Vec<String>,
 }
+
+crate enum StreamNext {
+    Last,
+    External,
+    Internal,
+}
+
+impl ExternalCommand {
+    crate async fn run(
+        self,
+        context: &mut Context,
+        mut input: ClassifiedInputStream,
+        stream_next: StreamNext,
+    ) -> Result<ClassifiedInputStream, ShellError> {
+        let mut process = Exec::shell(&self.name)
+            .args(&self.args)
+            .cwd(context.env.lock().unwrap().cwd());
+
+        let mut process = match stream_next {
+            StreamNext::Last => process,
+            StreamNext::External | StreamNext::Internal => {
+                process.stdout(subprocess::Redirection::Pipe)
+            }
+        };
+
+        if let Some(stdin) = input.stdin {
+            process = process.stdin(stdin);
+        }
+
+        let mut popen = process.popen().unwrap();
+
+        match stream_next {
+            StreamNext::Last => {
+                popen.wait()?;
+                Ok(ClassifiedInputStream::new())
+            }
+            StreamNext::External => {
+                let stdout = popen.stdout.take().unwrap();
+                Ok(ClassifiedInputStream::from_stdout(stdout))
+            }
+            StreamNext::Internal => {
+                let stdout = popen.stdout.take().unwrap();
+                let file = futures::io::AllowStdIo::new(stdout);
+                let stream = Framed::new(file, LinesCodec {});
+                let stream = stream.map(|line| Value::string(line.unwrap()));
+                Ok(ClassifiedInputStream::from_input_stream(stream.boxed()))
+            }
+        }
+
+        // if stream_next {
+        //     let stdout = popen.stdout.take().unwrap();
+        //     Ok(ClassifiedInputStream::from_stdout(stdout))
+        // } else {
+        //     // popen.stdin.take();
+        //     popen.wait()?;
+        //     Ok(ClassifiedInputStream::new())
+        // }
+    }
+}

src/commands/command.rs

@@ -1,25 +1,24 @@
 use crate::errors::ShellError;
 use crate::object::Value;
 use crate::prelude::*;
-use crate::Context;
 use std::path::PathBuf;
 
-pub struct CommandArgs<'caller> {
+pub struct CommandArgs {
-    pub host: &'caller mut dyn Host,
+    pub host: Arc<Mutex<dyn Host + Send>>,
-    pub env: &'caller crate::Environment,
+    pub env: Arc<Mutex<Environment>>,
     pub args: Vec<Value>,
-    pub input: VecDeque<Value>,
+    pub input: InputStream,
 }
 
-impl CommandArgs<'caller> {
+impl CommandArgs {
     crate fn from_context(
         ctx: &'caller mut Context,
         args: Vec<Value>,
-        input: VecDeque<Value>,
+        input: InputStream,
-    ) -> CommandArgs<'caller> {
+    ) -> CommandArgs {
         CommandArgs {
-            host: &mut ctx.host,
+            host: ctx.host.clone(),
-            env: &ctx.env,
+            env: ctx.env.clone(),
             args,
             input,
         }
@@ -38,26 +37,20 @@ pub enum ReturnValue {
 }
 
 impl ReturnValue {
-    crate fn single(value: Value) -> VecDeque<ReturnValue> {
-        let mut v = VecDeque::new();
-        v.push_back(ReturnValue::Value(value));
-        v
-    }
-
     crate fn change_cwd(path: PathBuf) -> ReturnValue {
         ReturnValue::Action(CommandAction::ChangeCwd(path))
     }
 }
 
 pub trait Command {
-    fn run(&self, args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError>;
+    fn run(&self, args: CommandArgs) -> Result<OutputStream, ShellError>;
 }
 
 impl<F> Command for F
 where
-    F: Fn(CommandArgs<'_>) -> Result<VecDeque<ReturnValue>, ShellError>,
+    F: Fn(CommandArgs) -> Result<OutputStream, ShellError>,
 {
-    fn run(&self, args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
+    fn run(&self, args: CommandArgs) -> Result<OutputStream, ShellError> {
         self(args)
     }
 }

src/commands/ls.rs

@@ -2,8 +2,8 @@ use crate::errors::ShellError;
 use crate::object::{dir_entry_dict, Value};
 use crate::prelude::*;
 
-pub fn ls(args: CommandArgs<'value>) -> Result<VecDeque<ReturnValue>, ShellError> {
+pub fn ls(args: CommandArgs) -> Result<OutputStream, ShellError> {
-    let cwd = args.env.cwd().to_path_buf();
+    let cwd = args.env.lock().unwrap().cwd().to_path_buf();
 
     let entries = std::fs::read_dir(&cwd).map_err(|e| ShellError::string(format!("{:?}", e)))?;
 
@@ -14,5 +14,5 @@ pub fn ls(args: CommandArgs<'value>) -> Result<VecDeque<ReturnValue>, ShellError
         shell_entries.push_back(ReturnValue::Value(value))
     }
 
-    Ok(shell_entries)
+    Ok(shell_entries.boxed())
 }

src/commands/ps.rs

@@ -4,7 +4,7 @@ use crate::object::Value;
 use crate::prelude::*;
 use sysinfo::SystemExt;
 
-pub fn ps(_args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
+pub fn ps(_args: CommandArgs) -> Result<OutputStream, ShellError> {
     let mut system = sysinfo::System::new();
     system.refresh_all();
 
@@ -15,5 +15,5 @@ pub fn ps(_args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellErr
         .map(|(_, process)| ReturnValue::Value(Value::Object(process_dict(process))))
         .collect::<VecDeque<_>>();
 
-    Ok(list)
+    Ok(list.boxed())
 }

src/commands/reject.rs

@@ -3,7 +3,7 @@ use crate::object::base::reject_fields;
 use crate::object::Value;
 use crate::prelude::*;
 
-pub fn reject(args: CommandArgs<'value>) -> Result<VecDeque<ReturnValue>, ShellError> {
+pub fn reject(args: CommandArgs) -> Result<OutputStream, ShellError> {
     if args.args.is_empty() {
         return Err(ShellError::string("select requires a field"));
     }
@@ -11,12 +11,10 @@ pub fn reject(args: CommandArgs<'value>) -> Result<VecDeque<ReturnValue>, ShellE
     let fields: Result<Vec<String>, _> = args.args.iter().map(|a| a.as_string()).collect();
     let fields = fields?;
 
-    let objects = args
+    let stream = args
         .input
-        .iter()
+        .map(move |item| Value::Object(reject_fields(&item, &fields)))
-        .map(|item| Value::Object(reject_fields(item, &fields)))
+        .map(|item| ReturnValue::Value(item));
-        .map(|item| ReturnValue::Value(item))
-        .collect();
 
-    Ok(objects)
+    Ok(stream.boxed())
 }

src/commands/select.rs

@@ -3,7 +3,7 @@ use crate::object::base::select_fields;
 use crate::object::Value;
 use crate::prelude::*;
 
-pub fn select(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
+pub fn select(args: CommandArgs) -> Result<OutputStream, ShellError> {
     if args.args.is_empty() {
         return Err(ShellError::string("select requires a field"));
     }
@@ -13,10 +13,9 @@ pub fn select(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, Shell
 
     let objects = args
         .input
-        .iter()
+        .map(move |item| Value::Object(select_fields(&item, &fields)))
-        .map(|item| Value::Object(select_fields(item, &fields)))
+        .map(|item| ReturnValue::Value(item));
-        .map(|item| ReturnValue::Value(item))
-        .collect();
 
-    Ok(objects)
+    let stream = Pin::new(Box::new(objects));
+    Ok(stream)
 }

src/commands/skip.rs

@@ -1,21 +1,13 @@
 use crate::errors::ShellError;
 use crate::prelude::*;
 
-pub fn skip(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
+pub fn skip(args: CommandArgs) -> Result<OutputStream, ShellError> {
     let amount = args.args[0].as_int()?;
 
-    let amount = if args.input.len() > amount as usize {
+    let input = args.input;
-        amount as usize
-    } else {
-        args.input.len()
-    };
 
-    let out: VecDeque<ReturnValue> = args
+    Ok(input
-        .input
+        .skip(amount as u64)
-        .into_iter()
-        .skip(amount)
         .map(|v| ReturnValue::Value(v))
-        .collect();
+        .boxed())
-
-    Ok(out)
 }

src/commands/sort_by.rs

@@ -1,23 +1,25 @@
 use crate::errors::ShellError;
 use crate::prelude::*;
 
-pub fn sort_by(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
+pub fn sort_by(args: CommandArgs) -> Result<OutputStream, ShellError> {
     let fields: Result<Vec<_>, _> = args.args.iter().map(|a| a.as_string()).collect();
     let fields = fields?;
 
-    let mut output = args.input.into_iter().collect::<Vec<_>>();
+    let output = args.input.collect::<Vec<_>>();
 
-    output.sort_by_key(|item| {
+    let output = output.map(move |mut vec| {
+        vec.sort_by_key(|item| {
             fields
                 .iter()
                 .map(|f| item.get_data_by_key(f).borrow().copy())
                 .collect::<Vec<Value>>()
         });
 
-    let output = output
+        vec.into_iter()
-        .iter()
+            .map(|v| ReturnValue::Value(v.copy()))
-        .map(|o| ReturnValue::Value(o.copy()))
+            .collect::<VecDeque<_>>()
-        .collect();
+            .boxed()
+    });
 
-    Ok(output)
+    Ok(output.flatten_stream().boxed())
 }

src/commands/take.rs

@@ -3,21 +3,13 @@ use crate::prelude::*;
 
 // TODO: "Amount remaining" wrapper
 
-pub fn take(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
+pub fn take(args: CommandArgs) -> Result<OutputStream, ShellError> {
     let amount = args.args[0].as_int()?;
 
-    let amount = if args.input.len() > amount as usize {
+    let input = args.input;
-        amount as usize
-    } else {
-        args.input.len()
-    };
 
-    let out: VecDeque<ReturnValue> = args
+    Ok(input
-        .input
+        .take(amount as u64)
-        .into_iter()
-        .take(amount)
         .map(|v| ReturnValue::Value(v))
-        .collect();
+        .boxed())
-
-    Ok(out)
 }

src/commands/to_array.rs

@@ -1,15 +1,17 @@
-use crate::errors::ShellError;
 use crate::object::Value;
 use crate::prelude::*;
 
-pub fn to_array(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
+pub fn to_array(args: CommandArgs) -> Result<OutputStream, ShellError> {
-    let out = args.input.into_iter().collect();
+    let out = args.input.collect();
-    Ok(ReturnValue::single(Value::List(out)))
+    Ok(out
+        .map(|vec: Vec<_>| single_output(Value::List(vec)))
+        .flatten_stream()
+        .boxed())
 }
 
-crate fn stream_to_array(stream: VecDeque<Value>) -> VecDeque<Value> {
+crate async fn stream_to_array(stream: InputStream) -> InputStream {
-    let out = Value::List(stream.into_iter().collect());
+    let out = Value::List(stream.collect().await);
     let mut stream = VecDeque::new();
     stream.push_back(out);
-    stream
+    stream.boxed() as InputStream
 }

src/commands/view.rs

@@ -2,14 +2,14 @@ use crate::errors::ShellError;
 use crate::prelude::*;
 use prettyprint::PrettyPrinter;
 
-pub fn view(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
+pub fn view(args: CommandArgs) -> Result<OutputStream, ShellError> {
     let target = match args.args.first() {
         // TODO: This needs better infra
         None => return Err(ShellError::string(format!("cat must take one arg"))),
         Some(v) => v.as_string()?.clone(),
     };
 
-    let cwd = args.env.cwd().to_path_buf();
+    let cwd = args.env.lock().unwrap().cwd().to_path_buf();
 
     let printer = PrettyPrinter::default()
         .line_numbers(false)
@@ -22,5 +22,5 @@ pub fn view(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellEr
 
     let _ = printer.file(file.display().to_string());
 
-    Ok(VecDeque::new())
+    Ok(VecDeque::new().boxed())
 }

src/commands/where_.rs

@@ -2,26 +2,27 @@ use crate::errors::ShellError;
 use crate::object::base::find;
 use crate::prelude::*;
 
-pub fn r#where(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
+pub fn r#where(args: CommandArgs) -> Result<OutputStream, ShellError> {
     if args.args.is_empty() {
         return Err(ShellError::string("select requires a field"));
     }
 
     let field: Result<String, _> = args.args[0].as_string();
     let field = field?;
+    let input = args.input;
+    let operator = args.args[1].copy();
 
-    match args.args[1] {
+    match operator {
-        Value::Primitive(Primitive::Operator(ref operator)) => {
+        Value::Primitive(Primitive::Operator(operator)) => {
-            let objects = args
+            let right = args.args[2].copy();
-                .input
-                .iter()
-                .filter(|item| find(&item, &field, operator, &args.args[2]))
-                .map(|item| ReturnValue::Value(item.copy()))
-                .collect();
 
-            Ok(objects)
+            let objects = input
+                .filter(move |item| futures::future::ready(find(&item, &field, &operator, &right)))
+                .map(|item| ReturnValue::Value(item.copy()));
+
+            Ok(objects.boxed())
         }
-        ref x => {
+        x => {
             println!("{:?}", x);
             Err(ShellError::string("expected a comparison operator"))
         }

src/context.rs

@@ -4,21 +4,21 @@ use std::error::Error;
 use std::sync::Arc;
 
 pub struct Context {
-    commands: indexmap::IndexMap<String, Arc<dyn crate::Command>>,
+    commands: indexmap::IndexMap<String, Arc<dyn Command>>,
-    crate host: Box<dyn crate::Host>,
+    crate host: Arc<Mutex<dyn Host + Send>>,
-    crate env: Environment,
+    crate env: Arc<Mutex<Environment>>,
 }
 
 impl Context {
     crate fn basic() -> Result<Context, Box<Error>> {
         Ok(Context {
             commands: indexmap::IndexMap::new(),
-            host: Box::new(crate::env::host::BasicHost),
+            host: Arc::new(Mutex::new(crate::env::host::BasicHost)),
-            env: crate::Environment::basic()?,
+            env: Arc::new(Mutex::new(Environment::basic()?)),
         })
     }
 
-    pub fn add_commands(&mut self, commands: Vec<(&str, Arc<dyn crate::Command>)>) {
+    pub fn add_commands(&mut self, commands: Vec<(&str, Arc<dyn Command>)>) {
         for (name, command) in commands {
             self.commands.insert(name.to_string(), command);
         }
@@ -36,11 +36,11 @@ impl Context {
         &mut self,
         command: Arc<dyn Command>,
         arg_list: Vec<Value>,
-        input: VecDeque<Value>,
+        input: InputStream,
-    ) -> Result<VecDeque<ReturnValue>, ShellError> {
+    ) -> Result<OutputStream, ShellError> {
         let command_args = CommandArgs {
-            host: &mut self.host,
+            host: self.host.clone(),
-            env: &self.env,
+            env: self.env.clone(),
             args: arg_list,
             input,
         };

src/env/host.rs

@@ -1,9 +1,20 @@
 pub trait Host {
+    fn out_terminal(&self) -> Box<term::StdoutTerminal>;
+    fn err_terminal(&self) -> Box<term::StderrTerminal>;
+
     fn stdout(&mut self, out: &str);
     fn stderr(&mut self, out: &str);
 }
 
 impl Host for Box<dyn Host> {
+    fn out_terminal(&self) -> Box<term::StdoutTerminal> {
+        (**self).out_terminal()
+    }
+
+    fn err_terminal(&self) -> Box<term::StderrTerminal> {
+        (**self).err_terminal()
+    }
+
     fn stdout(&mut self, out: &str) {
         (**self).stdout(out)
     }
@@ -16,11 +27,25 @@ impl Host for Box<dyn Host> {
 crate struct BasicHost;
 
 impl Host for BasicHost {
+    fn out_terminal(&self) -> Box<term::StdoutTerminal> {
+        term::stdout().unwrap()
+    }
+
+    fn err_terminal(&self) -> Box<term::StderrTerminal> {
+        term::stderr().unwrap()
+    }
+
     fn stdout(&mut self, out: &str) {
-        println!("{}", out)
+        match out {
+            "\n" => println!(""),
+            other => println!("{}", other),
+        }
     }
 
     fn stderr(&mut self, out: &str) {
-        eprintln!("{}", out)
+        match out {
+            "\n" => eprintln!(""),
+            other => eprintln!("{}", other),
+        }
     }
 }

src/errors.rs

@@ -1,7 +1,6 @@
 #[allow(unused)]
 use crate::prelude::*;
 
-use crate::Value;
 use derive_new::new;
 
 #[derive(Debug, Ord, PartialOrd, Eq, PartialEq, new)]
@@ -43,3 +42,21 @@ impl std::convert::From<std::io::Error> for ShellError {
         }
     }
 }
+
+impl std::convert::From<futures_sink::VecSinkError> for ShellError {
+    fn from(_input: futures_sink::VecSinkError) -> ShellError {
+        ShellError {
+            title: format!("Unexpected Vec Sink Error"),
+            error: Value::nothing(),
+        }
+    }
+}
+
+impl std::convert::From<subprocess::PopenError> for ShellError {
+    fn from(input: subprocess::PopenError) -> ShellError {
+        ShellError {
+            title: format!("{}", input),
+            error: Value::nothing(),
+        }
+    }
+}

src/format.rs

@@ -10,16 +10,9 @@ crate use generic::GenericView;
 crate use table::TableView;
 
 crate trait RenderView {
-    fn render_view(&self, host: &dyn Host) -> Vec<String>;
+    fn render_view(&self, host: &mut dyn Host) -> Result<(), ShellError>;
 }
 
-fn print_rendered(lines: &[String], host: &mut dyn Host) {
+crate fn print_view(view: &impl RenderView, host: &mut dyn Host) -> Result<(), ShellError> {
-    for line in lines {
+    view.render_view(host)
-        host.stdout(line);
-    }
-}
-
-crate fn print_view(view: &impl RenderView, host: &mut Host) {
-    // let mut ctx = context.lock().unwrap();
-    crate::format::print_rendered(&view.render_view(host), host);
 }

src/format/entries.rs

@@ -1,6 +1,5 @@
 use crate::format::RenderView;
 use crate::prelude::*;
-use crate::Host;
 
 use derive_new::new;
 
@@ -11,7 +10,7 @@ use derive_new::new;
 // another_name : ...
 #[derive(new)]
 pub struct EntriesView {
-    entries: Vec<(String, String)>,
+    entries: Vec<(crate::object::DescriptorName, String)>,
 }
 
 impl EntriesView {
@@ -32,17 +31,28 @@ impl EntriesView {
 }
 
 impl RenderView for EntriesView {
-    fn render_view(&self, _host: &dyn Host) -> Vec<String> {
+    fn render_view(&self, _host: &mut dyn Host) -> Result<(), ShellError> {
         if self.entries.len() == 0 {
-            return vec![];
+            return Ok(());
         }
 
-        let max_name_size: usize = self.entries.iter().map(|(n, _)| n.len()).max().unwrap();
+        let max_name_size: usize = self
-
+            .entries
-        self.entries
             .iter()
-            .map(|(k, v)| format!("{:width$} : {}", k, v, width = max_name_size))
+            .map(|(n, _)| n.display().len())
-            .collect()
+            .max()
+            .unwrap();
+
+        for (name, value) in &self.entries {
+            println!(
+                "{:width$} : {}",
+                name.display(),
+                value,
+                width = max_name_size
+            )
+        }
+
+        Ok(())
     }
 }
 
@@ -51,32 +61,30 @@ pub struct EntriesListView {
 }
 
 impl EntriesListView {
-    crate fn from_stream(values: VecDeque<Value>) -> EntriesListView {
+    crate async fn from_stream(values: InputStream) -> EntriesListView {
-        EntriesListView { values }
+        EntriesListView {
+            values: values.collect().await,
+        }
     }
 }
 
 impl RenderView for EntriesListView {
-    fn render_view(&self, host: &dyn Host) -> Vec<String> {
+    fn render_view(&self, host: &mut dyn Host) -> Result<(), ShellError> {
         if self.values.len() == 0 {
-            return vec![];
+            return Ok(());
         }
 
-        let mut strings = vec![];
-
         let last = self.values.len() - 1;
 
         for (i, item) in self.values.iter().enumerate() {
             let view = EntriesView::from_value(item);
-            let out = view.render_view(host);
+            view.render_view(host);
-
-            strings.extend(out);
 
             if i != last {
-                strings.push("\n".to_string());
+                host.stdout("\n");
             }
         }
 
-        strings
+        Ok(())
     }
 }

src/format/generic.rs

@@ -1,6 +1,6 @@
 use crate::format::{EntriesView, RenderView, TableView};
 use crate::object::Value;
-use crate::Host;
+use crate::prelude::*;
 use derive_new::new;
 
 // A list is printed one line at a time with an optional separator between groups
@@ -10,17 +10,17 @@ pub struct GenericView<'value> {
 }
 
 impl RenderView for GenericView<'value> {
-    fn render_view(&self, host: &dyn Host) -> Vec<String> {
+    fn render_view(&self, host: &mut dyn Host) -> Result<(), ShellError> {
         match self.value {
-            Value::Primitive(p) => vec![p.format(None)],
+            Value::Primitive(p) => Ok(host.stdout(&p.format(None))),
             Value::List(l) => {
                 let view = TableView::from_list(l);
 
                 if let Some(view) = view {
-                    view.render_view(host)
+                    view.render_view(host);
-                } else {
-                    vec![]
                 }
+
+                Ok(())
                 // let mut list: Vec<String> = vec![];
                 // for item in l {
                 //     match item {
@@ -39,11 +39,14 @@ impl RenderView for GenericView<'value> {
 
             o @ Value::Object(_) => {
                 let view = EntriesView::from_value(o);
-                let out = view.render_view(host);
+                view.render_view(host);
-                out
+                Ok(())
             }
 
-            Value::Error(e) => vec![format!("{}", e)],
+            Value::Error(e) => {
+                host.stdout(&format!("{:?}", e));
+                Ok(())
+            }
         }
     }
 }

src/format/list.rs

@@ -1,5 +1,5 @@
 use crate::format::RenderView;
-use crate::Host;
+use crate::prelude::*;
 use derive_new::new;
 
 // A list is printed one line at a time with an optional separator between groups
@@ -11,14 +11,12 @@ pub struct ListView {
 }
 
 impl RenderView for ListView {
-    fn render_view(&self, _host: &dyn Host) -> Vec<String> {
+    fn render_view(&self, host: &mut dyn Host) -> Result<(), ShellError> {
-        let mut out = vec![];
-
         for output in &self.list {
             let string: String = output.iter().map(|l| format!("{}\n", l)).collect();
-            out.push(format!("{}{}", string, self.sep));
+            host.stdout(&format!("{}{}", string, self.sep));
         }
 
-        out
+        Ok(())
     }
 }

src/format/table.rs

@@ -1,8 +1,9 @@
 use crate::format::RenderView;
 use crate::object::Value;
-use crate::Host;
+use crate::prelude::*;
+use ansi_term::Color;
 use derive_new::new;
-use prettytable::{Cell, Row, Table};
+use prettytable::{color, Attr, Cell, Row, Table};
 
 // An entries list is printed like this:
 //
@@ -24,7 +25,7 @@ impl TableView {
         let item = &values[0];
         let descs = item.data_descriptors();
 
-        let headers: Vec<String> = descs.iter().map(|d| d.name.clone()).collect();
+        let headers: Vec<String> = descs.iter().map(|d| d.name.display().to_string()).collect();
 
         let mut entries = vec![];
 
@@ -43,13 +44,27 @@ impl TableView {
 }
 
 impl RenderView for TableView {
-    fn render_view(&self, _host: &dyn Host) -> Vec<String> {
+    fn render_view(&self, host: &mut dyn Host) -> Result<(), ShellError> {
         if self.entries.len() == 0 {
-            return vec![];
+            return Ok(());
         }
 
         let mut table = Table::new();
-        let header: Vec<Cell> = self.headers.iter().map(|h| Cell::new(h)).collect();
+
+        // let format = prettytable::format::FormatBuilder::new();
+        // .column_separator(Color::Black.bold().paint("|"));
+
+        table.set_format(*prettytable::format::consts::FORMAT_NO_COLSEP);
+
+        let header: Vec<Cell> = self
+            .headers
+            .iter()
+            .map(|h| {
+                Cell::new(h)
+                    .with_style(Attr::ForegroundColor(color::GREEN))
+                    .with_style(Attr::Bold)
+            })
+            .collect();
 
         table.add_row(Row::new(header));
 
@@ -57,9 +72,8 @@ impl RenderView for TableView {
             table.add_row(Row::new(row.iter().map(|h| Cell::new(h)).collect()));
         }
 
-        let mut out = vec![];
+        table.print_term(&mut *host.out_terminal()).unwrap();
-        table.print(&mut out).unwrap();
 
-        vec![String::from_utf8_lossy(&out).to_string()]
+        Ok(())
     }
 }

src/main.rs

@@ -1,9 +1,10 @@
 #![feature(crate_visibility_modifier)]
 #![feature(in_band_lifetimes)]
+#![feature(async_await)]
+#![feature(try_trait)]
+#![feature(bind_by_move_pattern_guards)]
 
-#[allow(unused)]
+mod cli;
-use crate::prelude::*;
-
 mod commands;
 mod context;
 mod env;
@@ -13,243 +14,11 @@ mod object;
 mod parser;
 mod prelude;
 mod shell;
+mod stream;
 
-use crate::commands::classified::{ClassifiedCommand, ExternalCommand, InternalCommand};
-use crate::commands::command::ReturnValue;
-crate use crate::commands::command::{Command, CommandAction};
-use crate::context::Context;
-crate use crate::env::{Environment, Host};
-crate use crate::errors::ShellError;
-crate use crate::format::{EntriesListView, GenericView};
-use crate::object::Value;
-
-use rustyline::error::ReadlineError;
-use rustyline::{self, ColorMode, Config, Editor};
-use std::collections::VecDeque;
 use std::error::Error;
-use std::sync::Arc;
-
-#[derive(Debug)]
-pub enum MaybeOwned<'a, T> {
-    Owned(T),
-    Borrowed(&'a T),
-}
-
-impl<T> MaybeOwned<'a, T> {
-    crate fn borrow(&self) -> &T {
-        match self {
-            MaybeOwned::Owned(v) => v,
-            MaybeOwned::Borrowed(v) => v,
-        }
-    }
-}
 
 fn main() -> Result<(), Box<Error>> {
-    let config = Config::builder().color_mode(ColorMode::Forced).build();
+    futures::executor::block_on(crate::cli::cli())?;
-    let h = crate::shell::Helper::new();
-    let mut rl: Editor<crate::shell::Helper> = Editor::with_config(config);
-
-    #[cfg(windows)]
-    {
-        let _ = ansi_term::enable_ansi_support();
-    }
-
-    rl.set_helper(Some(h));
-    if rl.load_history("history.txt").is_err() {
-        println!("No previous history.");
-    }
-
-    let mut context = Context::basic()?;
-
-    {
-        use crate::commands::*;
-
-        context.add_commands(vec![
-            ("format", Arc::new(format)),
-            ("format-list", Arc::new(format_list)),
-            ("ps", Arc::new(ps::ps)),
-            ("ls", Arc::new(ls::ls)),
-            ("cd", Arc::new(cd::cd)),
-            ("view", Arc::new(view::view)),
-            ("skip", Arc::new(skip::skip)),
-            ("take", Arc::new(take::take)),
-            ("select", Arc::new(select::select)),
-            ("reject", Arc::new(reject::reject)),
-            ("to-array", Arc::new(to_array::to_array)),
-            ("where", Arc::new(where_::r#where)),
-            ("sort-by", Arc::new(sort_by::sort_by)),
-        ]);
-    }
-
-    loop {
-        let readline = rl.readline(&format!("{}> ", context.env.cwd().display().to_string()));
-
-        match process_line(readline, &mut context) {
-            LineResult::Success(line) => {
-                rl.add_history_entry(line.clone());
-            }
-
-            LineResult::Error(err) => {
-                context.host.stdout(&err);
-            }
-
-            LineResult::Break => {
-                break;
-            }
-
-            LineResult::FatalError(err) => {
-                context
-                    .host
-                    .stdout(&format!("A surprising fatal error occurred.\n{:?}", err));
-            }
-        }
-    }
-    rl.save_history("history.txt").unwrap();
-
     Ok(())
 }
-
-enum LineResult {
-    Success(String),
-    Error(String),
-    Break,
-
-    #[allow(unused)]
-    FatalError(ShellError),
-}
-
-fn process_line(readline: Result<String, ReadlineError>, ctx: &mut Context) -> LineResult {
-    match &readline {
-        Ok(line) if line.trim() == "exit" => LineResult::Break,
-
-        Ok(line) if line.trim() == "" => LineResult::Success(line.clone()),
-
-        Ok(line) => {
-            let result = match crate::parser::shell_parser(&line) {
-                Err(err) => {
-                    return LineResult::Error(format!("{:?}", err));
-                }
-
-                Ok(val) => val,
-            };
-
-            let parsed = result.1;
-
-            let mut input = VecDeque::new();
-
-            for item in parsed {
-                input = match process_command(item.clone(), input, ctx) {
-                    Ok(val) => val,
-                    Err(err) => return LineResult::Error(format!("{}", err.description())),
-                };
-            }
-
-            if input.len() > 0 {
-                if equal_shapes(&input) {
-                    let array = crate::commands::stream_to_array(input);
-                    let args = CommandArgs::from_context(ctx, vec![], array);
-                    match format(args) {
-                        Ok(_) => {}
-                        Err(err) => return LineResult::Error(err.to_string()),
-                    }
-                } else {
-                    let args = CommandArgs::from_context(ctx, vec![], input);
-                    match format(args) {
-                        Ok(_) => {}
-                        Err(err) => return LineResult::Error(err.to_string()),
-                    }
-                }
-            }
-
-            LineResult::Success(line.to_string())
-        }
-        Err(ReadlineError::Interrupted) => {
-            println!("CTRL-C");
-            LineResult::Break
-        }
-        Err(ReadlineError::Eof) => {
-            println!("CTRL-D");
-            LineResult::Break
-        }
-        Err(err) => {
-            println!("Error: {:?}", err);
-            LineResult::Break
-        }
-    }
-}
-
-fn process_command(
-    parsed: Vec<crate::parser::Item>,
-    input: VecDeque<Value>,
-    context: &mut Context,
-) -> Result<VecDeque<Value>, ShellError> {
-    let command = classify_command(&parsed, context)?;
-
-    command.run(input, context)
-}
-
-fn classify_command(
-    command: &[crate::parser::Item],
-    context: &Context,
-) -> Result<ClassifiedCommand, ShellError> {
-    let command_name = &command[0].name()?;
-
-    let arg_list: Vec<Value> = command[1..].iter().map(|i| i.as_value()).collect();
-    let arg_list_strings: Vec<String> = command[1..].iter().map(|i| i.print()).collect();
-
-    match *command_name {
-        other => match context.has_command(*command_name) {
-            true => {
-                let command = context.get_command(command_name);
-                Ok(ClassifiedCommand::Internal(InternalCommand {
-                    command,
-                    args: arg_list,
-                }))
-            }
-            false => Ok(ClassifiedCommand::External(ExternalCommand {
-                name: other.to_string(),
-                args: arg_list_strings,
-            })),
-        },
-    }
-}
-
-fn format(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
-    let last = args.input.len() - 1;
-    for (i, item) in args.input.iter().enumerate() {
-        let view = GenericView::new(item);
-        crate::format::print_view(&view, args.host);
-
-        if last != i {
-            println!("");
-        }
-    }
-
-    Ok(VecDeque::new())
-}
-
-fn format_list(args: CommandArgs<'caller>) -> Result<VecDeque<ReturnValue>, ShellError> {
-    let view = EntriesListView::from_stream(args.input);
-    crate::format::print_view(&view, args.host);
-
-    Ok(VecDeque::new())
-}
-
-fn equal_shapes(input: &VecDeque<Value>) -> bool {
-    let mut items = input.iter();
-
-    let item = match items.next() {
-        Some(item) => item,
-        None => return false,
-    };
-
-    let desc = item.data_descriptors();
-
-    for item in items {
-        if desc != item.data_descriptors() {
-            return false;
-        }
-    }
-
-    true
-}

src/object.rs

@@ -6,5 +6,6 @@ crate mod process;
 crate mod types;
 
 crate use base::{Primitive, Value};
+crate use desc::{DataDescriptor, DescriptorName};
 crate use dict::Dictionary;
 crate use files::dir_entry_dict;

src/object/base.rs

@@ -1,6 +1,7 @@
 use crate::errors::ShellError;
-use crate::object::desc::DataDescriptor;
+use crate::object::{DataDescriptor, DescriptorName};
 use crate::parser::parse::Operator;
+use crate::prelude::*;
 use ansi_term::Color;
 use chrono::{DateTime, Utc};
 use chrono_humanize::Humanize;
@@ -68,28 +69,28 @@ pub enum Value {
 impl Value {
     crate fn data_descriptors(&self) -> Vec<DataDescriptor> {
         match self {
-            Value::Primitive(_) => vec![],
+            Value::Primitive(_) => vec![DataDescriptor::value_of()],
             Value::Object(o) => o.data_descriptors(),
             Value::List(_) => vec![],
             Value::Error(_) => vec![],
         }
     }
 
-    crate fn get_data_by_key(&'a self, name: &str) -> crate::MaybeOwned<'a, Value> {
+    crate fn get_data_by_key(&'a self, name: &str) -> MaybeOwned<'a, Value> {
         match self {
-            Value::Primitive(_) => crate::MaybeOwned::Owned(Value::nothing()),
+            Value::Primitive(_) => MaybeOwned::Owned(Value::nothing()),
             Value::Object(o) => o.get_data_by_key(name),
-            Value::List(_) => crate::MaybeOwned::Owned(Value::nothing()),
+            Value::List(_) => MaybeOwned::Owned(Value::nothing()),
-            Value::Error(_) => crate::MaybeOwned::Owned(Value::nothing()),
+            Value::Error(_) => MaybeOwned::Owned(Value::nothing()),
         }
     }
 
-    crate fn get_data(&'a self, desc: &DataDescriptor) -> crate::MaybeOwned<'a, Value> {
+    crate fn get_data(&'a self, desc: &DataDescriptor) -> MaybeOwned<'a, Value> {
         match self {
-            Value::Primitive(_) => crate::MaybeOwned::Owned(Value::nothing()),
+            p @ Value::Primitive(_) => MaybeOwned::Borrowed(p),
             Value::Object(o) => o.get_data(desc),
-            Value::List(_) => crate::MaybeOwned::Owned(Value::nothing()),
+            Value::List(_) => MaybeOwned::Owned(Value::nothing()),
-            Value::Error(_) => crate::MaybeOwned::Owned(Value::nothing()),
+            Value::Error(_) => MaybeOwned::Owned(Value::nothing()),
         }
     }
 
@@ -202,9 +203,9 @@ crate fn select_fields(obj: &Value, fields: &[String]) -> crate::object::Diction
     let descs = obj.data_descriptors();
 
     for field in fields {
-        match descs.iter().find(|d| d.name == *field) {
+        match descs.iter().find(|d| d.name.is_string(field)) {
-            None => out.add(field.to_string(), Value::nothing()),
+            None => out.add(DataDescriptor::for_string_name(field), Value::nothing()),
-            Some(desc) => out.add(field.to_string(), obj.get_data(desc).borrow().copy()),
+            Some(desc) => out.add(desc.copy(), obj.get_data(desc).borrow().copy()),
         }
     }
 
@@ -217,10 +218,10 @@ crate fn reject_fields(obj: &Value, fields: &[String]) -> crate::object::Diction
     let descs = obj.data_descriptors();
 
     for desc in descs {
-        if fields.contains(&desc.name) {
+        match desc.name.as_string() {
-            continue;
+            None => continue,
-        } else {
+            Some(s) if fields.iter().any(|field| field == s) => continue,
-            out.add(desc.name.clone(), obj.get_data(&desc).borrow().copy())
+            Some(s) => out.add(desc.copy(), obj.get_data(&desc).borrow().copy()),
         }
     }
 
@@ -229,7 +230,7 @@ crate fn reject_fields(obj: &Value, fields: &[String]) -> crate::object::Diction
 
 crate fn find(obj: &Value, field: &str, op: &Operator, rhs: &Value) -> bool {
     let descs = obj.data_descriptors();
-    match descs.iter().find(|d| d.name == *field) {
+    match descs.iter().find(|d| d.name.is_string(field)) {
         None => false,
         Some(desc) => {
             let v = obj.get_data(desc).borrow().copy();

src/object/desc.rs

@@ -1,17 +1,110 @@
-use crate::object::types::Type;
+use crate::object::types::{AnyShell, Type};
 use derive_new::new;
 
-#[derive(new)]
+#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash)]
+pub enum DescriptorName {
+    String(String),
+    ValueOf,
+}
+
+impl DescriptorName {
+    crate fn display(&self) -> &str {
+        match self {
+            DescriptorName::String(s) => s,
+            DescriptorName::ValueOf => "value",
+        }
+    }
+
+    crate fn as_string(&self) -> Option<&str> {
+        match self {
+            DescriptorName::String(s) => Some(s),
+            DescriptorName::ValueOf => None,
+        }
+    }
+
+    crate fn is_string(&self, string: &str) -> bool {
+        match self {
+            DescriptorName::String(s) => s == string,
+            DescriptorName::ValueOf => false,
+        }
+    }
+}
+
+#[derive(Debug, new)]
 pub struct DataDescriptor {
-    crate name: String,
+    crate name: DescriptorName,
     crate readonly: bool,
     crate ty: Box<dyn Type>,
 }
 
+impl From<&str> for DataDescriptor {
+    fn from(input: &str) -> DataDescriptor {
+        DataDescriptor {
+            name: DescriptorName::String(input.to_string()),
+            readonly: true,
+            ty: Box::new(AnyShell),
+        }
+    }
+}
+
+impl From<String> for DataDescriptor {
+    fn from(input: String) -> DataDescriptor {
+        DataDescriptor {
+            name: DescriptorName::String(input),
+            readonly: true,
+            ty: Box::new(AnyShell),
+        }
+    }
+}
+
 impl PartialEq for DataDescriptor {
     fn eq(&self, other: &DataDescriptor) -> bool {
         self.name == other.name && self.readonly == other.readonly && self.ty.equal(&*other.ty)
     }
 }
 
-impl DataDescriptor {}
+impl std::hash::Hash for DataDescriptor {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.name.hash(state);
+        self.readonly.hash(state);
+        self.ty.id().hash(state);
+    }
+}
+
+impl Eq for DataDescriptor {}
+
+impl DescriptorName {
+    crate fn for_string_name(name: impl Into<String>) -> DescriptorName {
+        DescriptorName::String(name.into())
+    }
+}
+
+impl DataDescriptor {
+    crate fn value_of() -> DataDescriptor {
+        DataDescriptor {
+            name: DescriptorName::ValueOf,
+            readonly: true,
+            ty: Box::new(AnyShell),
+        }
+    }
+
+    crate fn for_name(name: impl Into<DescriptorName>) -> DataDescriptor {
+        DataDescriptor {
+            name: name.into(),
+            readonly: true,
+            ty: Box::new(AnyShell),
+        }
+    }
+
+    crate fn for_string_name(name: impl Into<String>) -> DataDescriptor {
+        DataDescriptor::for_name(DescriptorName::for_string_name(name))
+    }
+
+    crate fn copy(&self) -> DataDescriptor {
+        DataDescriptor {
+            name: self.name.clone(),
+            readonly: self.readonly,
+            ty: self.ty.copy(),
+        }
+    }
+}

src/object/dict.rs

@@ -1,15 +1,13 @@
-#[allow(unused)]
 use crate::prelude::*;
 
-use crate::object::desc::DataDescriptor;
+use crate::object::{DataDescriptor, DescriptorName};
 use crate::object::{Primitive, Value};
-use crate::MaybeOwned;
 use indexmap::IndexMap;
 use std::cmp::{Ordering, PartialOrd};
 
 #[derive(Debug, Default, Eq, PartialEq)]
 pub struct Dictionary {
-    entries: IndexMap<String, Value>,
+    entries: IndexMap<DataDescriptor, Value>,
 }
 
 impl PartialOrd for Dictionary {
@@ -43,7 +41,7 @@ impl PartialEq<Value> for Dictionary {
 }
 
 impl Dictionary {
-    crate fn add(&mut self, name: impl Into<String>, value: Value) {
+    crate fn add(&mut self, name: impl Into<DataDescriptor>, value: Value) {
         self.entries.insert(name.into(), value);
     }
 
@@ -51,31 +49,30 @@ impl Dictionary {
         let mut out = Dictionary::default();
 
         for (key, value) in self.entries.iter() {
-            out.add(key.clone(), value.copy());
+            out.add(key.copy(), value.copy());
         }
 
         out
     }
 
     crate fn data_descriptors(&self) -> Vec<DataDescriptor> {
-        self.entries
+        self.entries.iter().map(|(name, _)| name.copy()).collect()
-            .iter()
-            .map(|(name, _)| {
-                DataDescriptor::new(name.clone(), true, Box::new(crate::object::types::AnyShell))
-            })
-            .collect()
     }
 
     crate fn get_data(&'a self, desc: &DataDescriptor) -> MaybeOwned<'a, Value> {
-        match self.entries.get(&desc.name) {
+        match self.entries.get(desc) {
             Some(v) => MaybeOwned::Borrowed(v),
             None => MaybeOwned::Owned(Value::Primitive(Primitive::Nothing)),
         }
     }
 
     crate fn get_data_by_key(&self, name: &str) -> MaybeOwned<'_, Value> {
-        match self.entries.get(name) {
+        match self
-            Some(v) => MaybeOwned::Borrowed(v),
+            .entries
+            .iter()
+            .find(|(desc_name, _)| desc_name.name.is_string(name))
+        {
+            Some((_, v)) => MaybeOwned::Borrowed(v),
             None => MaybeOwned::Owned(Value::Primitive(Primitive::Nothing)),
         }
     }

src/object/types.rs

@@ -1,11 +1,14 @@
 use std::any::Any;
+use std::fmt::Debug;
 
-pub trait Type {
+pub trait Type: Debug + Send {
     fn as_any(&self) -> &dyn Any;
     fn equal(&self, other: &dyn Type) -> bool;
+    fn id(&self) -> u64;
+    fn copy(&self) -> Box<Type>;
 }
 
-#[derive(Eq, PartialEq)]
+#[derive(Debug, Eq, PartialEq)]
 pub struct AnyShell;
 
 impl Type for AnyShell {
@@ -16,4 +19,12 @@ impl Type for AnyShell {
     fn equal(&self, other: &dyn Type) -> bool {
         other.as_any().is::<AnyShell>()
     }
+
+    fn id(&self) -> u64 {
+        0
+    }
+
+    fn copy(&self) -> Box<Type> {
+        Box::new(AnyShell)
+    }
 }

src/parser/completer.rs

@@ -1,9 +1,10 @@
+use crate::prelude::*;
 use rustyline::{completion, Context};
 use std::collections::BTreeMap;
 
 #[allow(unused)]
 crate struct Completer {
-    commands: BTreeMap<String, Box<dyn crate::Command>>,
+    commands: BTreeMap<String, Box<dyn Command>>,
 }
 
 impl completion::Completer for Completer {

src/parser/parse.rs

@@ -17,7 +17,7 @@ pub enum Item {
     Operator(Operator),
 }
 
-#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
 pub enum Operator {
     Equal,
     NotEqual,

src/prelude.rs

@@ -1,5 +1,13 @@
-crate use crate::commands::command::{Command, CommandArgs, ReturnValue};
+crate use crate::cli::MaybeOwned;
+crate use crate::commands::command::{Command, CommandAction, CommandArgs, ReturnValue};
+crate use crate::context::Context;
 crate use crate::env::{Environment, Host};
 crate use crate::errors::ShellError;
 crate use crate::object::{Primitive, Value};
+#[allow(unused)]
+crate use crate::stream::{empty_stream, single_output, InputStream, OutputStream};
+#[allow(unused)]
+crate use futures::{Future, FutureExt, Sink, SinkExt, Stream, StreamExt};
 crate use std::collections::VecDeque;
+crate use std::pin::Pin;
+crate use std::sync::{Arc, Mutex};

src/stream.rs

@@ -0,0 +1,16 @@
+use crate::prelude::*;
+use futures::stream::BoxStream;
+
+pub type InputStream = BoxStream<'static, Value>;
+pub type OutputStream = BoxStream<'static, ReturnValue>;
+
+crate fn empty_stream() -> OutputStream {
+    VecDeque::new().boxed()
+}
+
+crate fn single_output(item: Value) -> OutputStream {
+    let value = ReturnValue::Value(item);
+    let mut vec = VecDeque::new();
+    vec.push_back(value);
+    vec.boxed()
+}