(self, x, metadata=None)
return [element for element in result if element is not None]
Push data into the stream at this point

        Parameters
        ----------
        x: any
            an element of data
        metadata: list[dict], optional
            Various types of metadata associated with the data element in `x`.

            ref: RefCounter
            A reference counter used to check when data is done
Push data into the stream at this point
Push data into the stream at this point
def _emit(self, x, metadata=None):
        """
        Push data into the stream at this point

        Parameters
        ----------
        x: any
            an element of data
        metadata: list[dict], optional
            Various types of metadata associated with the data element in `x`.

            ref: RefCounter
            A reference counter used to check when data is done

        """
        self.current_value = x
        self.current_metadata = metadata
        if metadata:
            self._retain_refs(metadata, len(self.downstreams))
        else:
            metadata = []

        result = []
        for downstream in list(self.downstreams):
            r = downstream.update(x, who=self, metadata=metadata)

            if type(r) is list:
                result.extend(r)
            else:
                result.append(r)

            self._release_refs(metadata)

        return [element for element in result if element is not None]
def _emit ( self , x , metadata = None ) : self . current_value = x self . current_metadata = metadata if metadata : self . _retain_refs ( metadata , len ( self . downstreams ) ) else : metadata = [ ] result = [ ] for downstream in list ( self . downstreams ) : r = downstream . update ( x , who = self , metadata = metadata ) if type ( r ) is list : result . extend ( r ) else : result . append ( r ) self . _release_refs ( metadata ) return [ element for element in result if element is not None ]