searchc.cpp

Windows CE 6.0 Server 源码

        if (strcmp(pMessageItem->pSsdpMessage->szUSN, szUSN) == 0)
        {
            LeaveCriticalSection(&SearchRequest->cs);
            return TRUE;
        }
        p = p->Flink;
    }
    LeaveCriticalSection(&SearchRequest->cs);
    return FALSE;
}

BOOL CallbackOnSearchResult(PSSDP_SEARCH_REQUEST SearchRequest,
                            PSSDP_MESSAGE pSsdpMessage)
{
    EnterCriticalSection(&SearchRequest->cs);

    if(SearchRequest->state != SEARCH_COMPLETED)
    {
        InterlockedIncrement(&SearchRequest->bCallbackInProgress);

        LeaveCriticalSection(&SearchRequest->cs);

        SearchRequest->Callback(SSDP_FOUND, pSsdpMessage ,SearchRequest->Context);

        InterlockedDecrement(&SearchRequest->bCallbackInProgress);
    }
    else
        LeaveCriticalSection(&SearchRequest->cs);

    return TRUE;
}



static void
SearchCallback(
    SSDP_CALLBACK_TYPE CallbackType,    // ignored
    CONST SSDP_MESSAGE *pSsdpMessage,
    void *pContext)
{
    PSSDP_SEARCH_REQUEST SearchRequest = (PSSDP_SEARCH_REQUEST)pContext;

    EnterCriticalSection(&SearchRequest->cs);
    if(++SearchRequest->dwNumberOfResponses <= upnp_config::max_search_results())
        AddRequestToListResponse(&SearchRequest->ListResponses, pSsdpMessage);
    LeaveCriticalSection(&SearchRequest->cs);
}


DWORD WINAPI DoSsdpSearchThread(LPVOID lpvThreadParam)
{
    HINSTANCE hLib = LoadLibrary(L"upnpctrl.dll");

    CoInitializeEx(NULL, COINIT_MULTITHREADED);

    PSSDP_SEARCH_REQUEST SearchRequest = (PSSDP_SEARCH_REQUEST) lpvThreadParam;

    DWORD dwRet = DoSsdpSearch(SearchRequest);

    FreeSearchRequest(SearchRequest);

    CoUninitialize();

    FreeLibraryAndExitThread(hLib, dwRet);

    return dwRet;
}


DWORD WINAPI DoSsdpSearch(PSSDP_SEARCH_REQUEST SearchRequest)
{
    // Get cache content put it in the list response.

    fd_set ReadSocketSet;
    struct timeval SelectTimeOut;
    INT nRet;

    // In case the loopback packet failed, this is our last resort.
    SelectTimeOut.tv_sec = SELECT_TIMEOUT;
    SelectTimeOut.tv_usec = 0;

    // It is OK to check fExit with CSListSearch, as the memory will not be
    // freed until this thread exits.
    while (InterlockedExchange(&SearchRequest->fExit, SearchRequest->fExit) == 0)
    {
        CHAR *RecvBuf = NULL;
        SOCKADDR_STORAGE RemoteSocket;
        SSDP_REQUEST SsdpResponse;
        INT SocketSize = sizeof(RemoteSocket);
        unsigned int i;

        FD_ZERO(&ReadSocketSet);

        EnterCriticalSection(&SearchRequest->cs);

        FD_SET(SearchRequest->WakeupSocket, &ReadSocketSet);

        for(ce::vector<SSDP_SEARCH_NETWORK>::iterator it = SearchRequest->Networks.begin(), itEnd = SearchRequest->Networks.end(); it != itEnd; ++it)
            FD_SET(it->socket, &ReadSocketSet);

        LeaveCriticalSection(&SearchRequest->cs);

        nRet = select(-1, &ReadSocketSet, NULL, NULL, &SelectTimeOut);

        if (nRet == SOCKET_ERROR)
        {
            TraceTag(ttidError, "select failed with error %d",
                     GetLastError());
            break;
        }

        if (nRet == 0)
        {
            TraceTag(ttidError, "!!! select timed out !!!, where is "
                     "loopback packet? ");
            break;
        }

        for (i = 0; i < ReadSocketSet.fd_count; i++)
        {
            if (TRUE == SocketReceive(ReadSocketSet.fd_array[i], &RecvBuf, &RemoteSocket))
            {
                InitializeSsdpRequest(&SsdpResponse);

                if (ParseSsdpResponse(RecvBuf, &SsdpResponse) == TRUE)
                {
                    DWORD dw;
                    
                    if(SsdpResponse.Headers[SSDP_LOCATION])
                    {
                        DWORD dwIndex = 0;
                        
                        // find the network on which the response arrived
                        for(ce::vector<SSDP_SEARCH_NETWORK>::iterator it = SearchRequest->Networks.begin(), itEnd = SearchRequest->Networks.end(); it != itEnd; ++it)
                            if(it->socket == ReadSocketSet.fd_array[i])
                            {
                                dwIndex = it->dwIndex;
                                break;
                            }

                        Assert(dwIndex != 0);

                        if(!FixURLAddressScopeId(SsdpResponse.Headers[SSDP_LOCATION], dwIndex, NULL, &(dw = 0)) && dw != 0)
                            if(char* pszLocation = (char*)SsdpAlloc(dw))
                                if(FixURLAddressScopeId(SsdpResponse.Headers[SSDP_LOCATION], dwIndex, pszLocation, &dw))
                                {
                                    SsdpFree(SsdpResponse.Headers[SSDP_LOCATION]);
                                    SsdpResponse.Headers[SSDP_LOCATION] = pszLocation;
                                }
                                else
                                    SsdpFree(pszLocation);
                    }

                    // Check duplicate
                    PSSDP_MESSAGE pSsdpMessage;
                    PSSDP_MESSAGE_ITEM pMessageItem;

                    if (!IsInTheListResponse(SearchRequest,
                                             SsdpResponse.Headers[SSDP_USN]))
                    {
                        pMessageItem = (PSSDP_MESSAGE_ITEM) malloc(sizeof(SSDP_MESSAGE_ITEM));
                        if (pMessageItem == NULL)
                        {
                            TraceTag(ttidError, "Error in recvfrom. Code (%d)",
                                     WSAGetLastError());
                        }
                        else
                        {

                            if (pSsdpMessage = InitializeSsdpMessageFromRequest(&SsdpResponse) )
                            {
                                pMessageItem->pSsdpMessage = pSsdpMessage;

                                if (SearchRequest->Callback != NULL)
                                {
                                    CallbackOnSearchResult(SearchRequest,
                                                           pSsdpMessage);
                                }

                                EnterCriticalSection(&SearchRequest->cs);
                                InsertHeadList(&SearchRequest->ListResponses,
                                               &(pMessageItem->linkage));
                                LeaveCriticalSection(&SearchRequest->cs);
                                // Update the server cache

                                //UpdateCache(pSsdpMessage);
                            }
                            else
                            {
                                free(pMessageItem);
                            }
                        }


                    }

                    FreeSsdpRequest(&SsdpResponse);
                }

                free(RecvBuf);
            }
        }
    }

    EnterCriticalSection(&SearchRequest->cs);

    SocketClose(SearchRequest->WakeupSocket);
    CloseSearchSockets(SearchRequest);

    if (SearchRequest->Callback != NULL && SearchRequest->state != SEARCH_COMPLETED)
    {
        InterlockedIncrement(&SearchRequest->bCallbackInProgress); 

        LeaveCriticalSection(&SearchRequest->cs);

        SearchRequest->Callback(SSDP_DONE, NULL, SearchRequest->Context);

        InterlockedDecrement(&SearchRequest->bCallbackInProgress); 
    }
    else
        LeaveCriticalSection(&SearchRequest->cs);

    return 0;
}



DWORD WINAPI SearchRequestTimerProc (VOID *Arg)
{
    PSSDP_SEARCH_REQUEST SearchRequest = (PSSDP_SEARCH_REQUEST) Arg;

    TraceTag(ttidSsdpSearch, "Entering Critical Section %x", CSListSearch);
    EnterCriticalSection(&SearchRequest->cs);

    TraceTag(ttidSsdpSearch, "Search request timer of %x expired with count = %d",
             SearchRequest,SearchRequest->NumOfRetry);

    if (InterlockedExchange(&SearchRequest->fExit, SearchRequest->fExit) == 1)
    {
        SetEvent(SearchRequest->DoneEvent);
        LeaveCriticalSection(&SearchRequest->cs);
        return 0;
    }

    _try
    {
        if (SearchRequest->NumOfRetry == 0)
        {
            TraceTag(ttidSsdpSearchResp, "The last search request timed out"
                     ".%x ", SearchRequest);
            InterlockedIncrement(&SearchRequest->fExit);
            WakeupSelect(SearchRequest);
            SetEvent(SearchRequest->DoneEvent);
        }
        else
        {
            SendSearchRequest(SearchRequest);

            SearchRequest->NumOfRetry--;

            SearchRequest->dwTimerCookie = SearchRequest->ThreadPool.StartTimer(SearchRequestTimerProc, (VOID *) SearchRequest, MX_VALUE);

            if (SearchRequest->dwTimerCookie == 0)
            {
                TraceTag(ttidError, "Failed to start search resp. timer.");
                InterlockedIncrement(&SearchRequest->fExit);
                WakeupSelect(SearchRequest);
                SetEvent(SearchRequest->DoneEvent);
            }
            else
            {
                TraceTag(ttidSsdpSearchResp, "Started search request timer.");
            }
        }
    }
    _finally
    {
        TraceTag(ttidSsdpSearchResp, "Leaving Critical Section %x", CSListSearch);

        LeaveCriticalSection(&SearchRequest->cs);
    }

    return 0;
}



VOID StopSearchRequestTimer(PSSDP_SEARCH_REQUEST SearchRequest)
{
    EnterCriticalSection(&SearchRequest->cs);

    // CSListSearch->fExit should be 1

    TraceTag(ttidSsdpAnnounce, "Stopping Search Request timer for %x",
             SearchRequest);

    if (SearchRequest->ThreadPool.StopTimer(SearchRequest->dwTimerCookie))
    {
        TraceTag(ttidSsdpAnnounce, "Search request timer stopped for service "
                 "%x", SearchRequest);
        LeaveCriticalSection(&SearchRequest->cs);
    }
    else
    {
        // Assume CTEStartTimer returned before FindServiceCallback returned.
        // Timer is running, wait for DoneEvent
        TraceTag(ttidSsdpAnnounce, "Search request timer is running, wait "
                 "...%x", SearchRequest);
        LeaveCriticalSection(&SearchRequest->cs);
        WaitForSingleObject(SearchRequest->DoneEvent, INFINITE);
    }
}



VOID WakeupSelect(PSSDP_SEARCH_REQUEST SearchRequest)
{
    SOCKADDR_IN  SockAddr;
    INT AddrSize = sizeof( SockAddr);

    ASSERT(InterlockedExchange(&SearchRequest->fExit, SearchRequest->fExit) == 1);

    if ( getsockname(SearchRequest->WakeupSocket,
                     (struct sockaddr *) &SockAddr,
                     &AddrSize) != SOCKET_ERROR)
    {
        SockAddr.sin_addr.s_addr = inet_addr("127.0.0.1");
        TraceTag(ttidSsdpSearch, "Sending to 127.0.0.1:%d",
                 ntohs(SockAddr.sin_port));
        // Will select get this if the socket is not bound to ADDR_ANY?
        SocketSend("Q", SearchRequest->WakeupSocket, (PSOCKADDR)&SockAddr);
    }
    else
    {
        TraceTag(ttidError, "Failed to send loopback packet. (%d)",
                 GetLastError());
        // select will eventually timeout, just slower.
    }
}