}

    /**
     * Manually unblock an IP
     *
     * @param sourceIp the IP to unblock
     * @return true if IP was unblocked, false if not found
     */
    public boolean unblockIp(String sourceIp) {
        IpAttempts ip = ipAttempts.get(sourceIp);
        if (ip != null) {
            ip.reset();
            log.info("Manually unblocked IP: {}", sourceIp);
            return true;
        }

        assertFalse(rateLimiter.checkAttempt("user6", ip), "IP should be blocked");

        // Manually unblock
        assertTrue(rateLimiter.unblockIp(ip), "Manual unblock should succeed");

        // Should be allowed again
        assertTrue(rateLimiter.checkAttempt("user7", ip), "Should be allowed after manual unblock");
    }

    @Test
    public void testNullUsername() throws Exception {

	m.acked = true
}

func (m *muxClient) unblockSend(seq uint32) {
	if !m.checkSeq(seq) {
		return
	}
	select {
	case m.outBlock <- struct{}{}:
	default:
		gridLogIf(m.ctx, errors.New("output unblocked overflow"))
	}
}

func (m *muxClient) pong(msg pongMsg) {
	if msg.NotFound || msg.Err != nil {
		err := errors.New("remote terminated call")
		if msg.Err != nil {

	}
	m.recvMu.Lock()
	defer m.recvMu.Unlock()
	if m.outBlock == nil {
		// Closed
		return
	}
	select {
	case m.outBlock <- struct{}{}:
	default:
		gridLogIf(m.ctx, errors.New("output unblocked overflow"))
	}
}

func (m *muxServer) ping(seq uint32) pongMsg {
	if !m.checkSeq(seq) {
		msg := fmt.Sprintf("receive sequence number mismatch. want %d, got %d", m.RecvSeq, seq)
		return pongMsg{Err: &msg}

	// OpMuxServerMsg contains a message to a server Mux
	OpMuxServerMsg

	// OpUnblockSrvMux contains a message that a server mux is unblocked with one.
	// Only Stateful streams has flow control.
	OpUnblockSrvMux

	// OpUnblockClMux contains a message that a client mux is unblocked with one.
	// Only Stateful streams has flow control.
	OpUnblockClMux

	// OpAckMux acknowledges a mux was created.
	OpAckMux

	}
}

// IsSet will return whether the singleton has been set.
func (s *Singleton[T]) IsSet() bool {
	select {
	case <-s.set:
		return true
	default:
		return false
	}
}

// Set the value and unblock all Get requests.
// This may only be called once, a second call will panic.
func (s *Singleton[T]) Set(v *T) {
	s.v = v
	close(s.set)

	PutByteBuffer(m.Payload)
	m.Payload = nil
	v, ok := c.outgoing.Load(m.MuxID)
	if !ok {
		if debugPrint {
			fmt.Println(c.Local, "Unblock: Unknown Mux:", m.MuxID)
		}
		// We can expect to receive unblocks for closed muxes
		return
	}
	v.unblockSend(m.Seq)
}

func (c *Connection) handleUnblockSrvMux(m message) {
	if m.Payload != nil {
		PutByteBuffer(m.Payload)

		t.Error("unexpected unblocking")
	case <-time.After(100 * time.Millisecond):
	}

	// unlock
	if err = dupl.Close(); err != nil {
		t.Fatal(err)
	}

	// the previously blocked routine should be unblocked
	select {
	case <-locked:
	case <-time.After(1 * time.Second):
		t.Error("unexpected blocking")
	}

    assertTrue(task.isDone());
    assertFalse(task.isCancelled());
  }

  public void testListenerCalledOnException() throws Exception {
    throwException = true;

    // Start up the task and unblock the latch to finish the task.
    exec.execute(task);
    runLatch.await();
    taskLatch.countDown();

    ExecutionException e = assertThrows(ExecutionException.class, () -> task.get(5, SECONDS));

    assertTrue(task.isDone());
    assertFalse(task.isCancelled());
  }

  public void testListenerCalledOnException() throws Exception {
    throwException = true;

    // Start up the task and unblock the latch to finish the task.
    exec.execute(task);
    runLatch.await();
    taskLatch.countDown();

    ExecutionException e = assertThrows(ExecutionException.class, () -> task.get(5, SECONDS));