CombineRex : Unexpected PublisherType behavior about swiftrex HOT 3 CLOSED

import Combine
import Dispatch
import PlaygroundSupport
import SwiftUI

let subject1 = PassthroughSubject<Int, Never>()

struct ContentView: View {
    @State var accumulated: [String] = []
    @State var last: Int = 0

    var body: some View {
        VStack {
            Button("Tap") {
                self.last += 1
                subject1.send(self.last)
            }

            Text(accumulated.joined(separator: ", "))
        }.onReceive(
            subject1
            //.removeDuplicates()
            .throttle(
                for: .milliseconds(500),
                scheduler: DispatchQueue.main,
                latest: true
            )
            .handleEvents(receiveCancel: {
                print("cancel")
            })
            //.removeDuplicates()
            //.eraseToAnyPublisher()
        ) { value in
            self.accumulated += ["\(value)"]
        }
    }
}

PlaygroundPage.current.liveView = UIHostingController(rootView: ContentView())

import Foundation

extension Publisher {
    public func removeDuplicatesEx(by predicate: @escaping (Output, Output) -> Bool) -> Publishers.RemoveDuplicatesEx<Self> {
        return .init(upstream: self, predicate: predicate)
    }
}
extension Publishers {
    public struct RemoveDuplicatesEx<Upstream: Publisher>: Publisher {
        public typealias Output = Upstream.Output
        public typealias Failure = Upstream.Failure
        public let upstream: Upstream
        public let predicate: (Upstream.Output, Upstream.Output) -> Bool

        public init(upstream: Upstream, predicate: @escaping (Publishers.RemoveDuplicatesEx<Upstream>.Output, Publishers.RemoveDuplicatesEx<Upstream>.Output) -> Bool) {
            self.upstream = upstream
            self.predicate = predicate
        }
        
        public func receive<S: Subscriber>(subscriber: S) where Upstream.Output == S.Input, S.Failure == Publishers.RemoveDuplicatesEx<Upstream>.Failure {
            let s = Inner(predicate: predicate, upstream: upstream, downstream: subscriber)
            subscriber.receive(subscription: s)
        }
    }
}

extension Publishers.RemoveDuplicatesEx {
    private final class Inner<Subscriber: Combine.Subscriber>: Combine.Subscription
    where
        Subscriber.Input == Output,
        Subscriber.Failure == Failure {
        
        typealias Input = Upstream.Output
        typealias Failure = Upstream.Failure
        
        typealias Predicate = (Upstream.Output, Upstream.Output) -> Bool
        
        let predicate: Predicate
        let downstream: Subscriber
        var upstreamSubscription: AnyCancellable?
        var previous: Output?
        
        init(predicate: @escaping (Output, Output) -> Bool, upstream: Upstream, downstream: Subscriber) {
            self.predicate = predicate
            self.downstream = downstream
            upstreamSubscription = upstream.sink(receiveCompletion: { completion in
                self.receive(completion: completion)
            }, receiveValue: { output in
                self.receive(output)
            })
        }
        
        func cancel() {
        }
        
        func request(_ demand: Subscribers.Demand) {
        }

        func receive(_ input: Input) -> Subscribers.Demand {
            guard let previous = self.previous else {
                self.previous = input
                _ = self.downstream.receive(input)
                return .unlimited
            }
            if self.predicate(previous, input) {
                self.previous = input
                _ = self.downstream.receive(input)
            }
            return .unlimited
        }
        
        func receive(completion: Subscribers.Completion<Failure>) {
            self.downstream.receive(completion: completion)
        }
    }
}