Handle downstream reduced in a multiplexed complete

[zmthy]

Fixes two related bugs I found while trying to write my own transducer with multiplexed reducing functions. Two of the transducers that use multiplexed do not correctly handle the potential for a reduced value being returned from the complete of an xform around a multiplexed reducing function when the complete invokes the step of the multiplexed function. Consider the body of the complete for the reduce transducer:

(let-complete [f-acc vacc]
  (rf (unreduced (rf acc (f (unreduced f-acc)))))))

reduce applies the step of rf and then takes care to unwrap any reduced value before passing it to the complete of rf. If rf is multiplexed, then its complete is identity, and a reduced value from the step is wrapped in two levels of reduced, resulting in a reduced value being returned from this complete of reduce. This is the correct behaviour, but by-key and multiplex do not correctly respect the resulting reduced value.

The first resulting bug is just in multiplex: if the complete of a multiplexed function returns a reduced value, then the complete will be invoked again. This is because the function that handles invoking all of the reducing functions always completes with the current function if invoking it returns a reduced value, regardless of whether the original invocation was also a completion. The first commit fixes this by adding an extra parameter to indicate if the current invocation is a step or a complete. I've added a test for this bug in the form of the following expression (plus one more for the map case):

(transduce
  (x/multiplex [(x/reduce rf/last)])
  rf/some
  [3 5 2])

This currently returns nil because let-complete in the complete of x/reduce has voided its state, causing it to have no value when it is completed a second time. The first commit includes a test that the result is now 2.

The second bug is in both by-key and multiplex: if the downstream reducing function returns a reduced value in the call to a multiplexed's complete, no more inputs will be processed but all of the remaining multiplexed functions will still be completed when the entire reduction should be stopped and only the original reducing function should be completed. The second commit fixes this by testing if the result of a multiplexed completion is a reduced value after its step returns a reduced value, and destroying any remaining reducing functions in the state if it is. I've added a test for this bug in the form of the following expression (plus two more for the multiplex cases):

(transduce
  (x/by-key (x/reduce (fn
                        ([] 0)
                        ([sum] sum)
                        ([sum x]
                         (let [sum (+ sum x)]
                           (if (> sum 4) (reduced 4) sum))))))
  rf/some
  [[:x 3] [:y 5] [:x 2]])

This currently returns [:x 3] because the :y case bails out of the entire computation when it stops its multiplexed invocation and discovers that the downstream wants to stop too, but then the :x case is still completed without having processed any more of its inputs. The second commit includes a test that the result is now [:y 4].