import com.mpush.api.Constants; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public void handle(HttpExchange httpExchange) throws IOException {
    String body = new String(readBody(httpExchange), Constants.UTF_8);
    Map<String, Object> params = Jsons.fromJson(body, Map.class);

    sendPush(params);

    byte[] data = "服务已经开始推送,请注意查收消息".getBytes(Constants.UTF_8);
    httpExchange.getResponseHeaders().set("Content-Type", "text/plain; charset=utf-8");
    httpExchange.sendResponseHeaders(200, data.length);//200, content-length
    OutputStream out = httpExchange.getResponseBody();
    out.write(data);
    out.close();
    httpExchange.close();
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
public static PushRequest build(MPushClient mPushClient, PushContext ctx) {
    byte[] content = ctx.getContext();
    PushMsg msg = ctx.getPushMsg();
    if (msg != null) {
        String json = Jsons.toJson(msg);
        if (json != null) {
            content = json.getBytes(Constants.UTF_8);
        }
    }

    Objects.requireNonNull(content, "push content can not be null.");

    return new PushRequest(mPushClient)
            .setAckModel(ctx.getAckModel())
            .setUserId(ctx.getUserId())
            .setTags(ctx.getTags())
            .setCondition(ctx.getCondition())
            .setTaskId(ctx.getTaskId())
            .setContent(content)
            .setTimeout(ctx.getTimeout())
            .setCallback(ctx.getCallback());

}
 

import com.mpush.api.Constants; //导入依赖的package包/类
public void kickUser(String userId, int clientType) {
    Set<RemoteRouter> remoteRouters = remoteRouterManager.lookupAll(userId);
    if (remoteRouters != null) {
        for (RemoteRouter remoteRouter : remoteRouters) {
            ClientLocation location = remoteRouter.getRouteValue();
            if (clientType == -1 || location.getClientType() == clientType) {
                MQKickRemoteMsg message = new MQKickRemoteMsg()
                        .setUserId(userId)
                        .setClientType(location.getClientType())
                        .setConnId(location.getConnId())
                        .setDeviceId(location.getDeviceId())
                        .setTargetServer(location.getHost())
                        .setTargetPort(location.getPort());
                mqClient.publish(Constants.getKickChannel(location.getHostAndPort()), message);
            }
        }
    }
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
private static void test() {
    Pair<RSAPublicKey, RSAPrivateKey> pair = RSAUtils.genKeyPair(RAS_KEY_SIZE);
    //生成公钥和私钥
    RSAPublicKey publicKey = pair.key;
    RSAPrivateKey privateKey = pair.value;
    //模
    String modulus = publicKey.getModulus().toString();
    //公钥指数
    String public_exponent = publicKey.getPublicExponent().toString();
    //私钥指数
    String private_exponent = privateKey.getPrivateExponent().toString();
    //明文
    byte[] ming = "123456789".getBytes(Constants.UTF_8);
    System.out.println("明文：" + new String(ming, Constants.UTF_8));
    //使用模和指数生成公钥和私钥
    RSAPrivateKey priKey = RSAUtils.getPrivateKey(modulus, private_exponent);
    RSAPublicKey pubKey = RSAUtils.getPublicKey(modulus, public_exponent);
    System.out.println("privateKey=" + priKey);
    System.out.println("publicKey=" + pubKey);
    //加密后的密文
    byte[] mi = RSAUtils.encryptByPublicKey(ming, pubKey);
    System.out.println("密文：" + new String(mi, Constants.UTF_8));
    //解密后的明文
    ming = RSAUtils.decryptByPrivateKey(mi, priKey);
    System.out.println("解密：" + new String(ming, Constants.UTF_8));
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
    int keySize = RAS_KEY_SIZE;
    if (args.length > 0) keySize = Integer.parseInt(args[0]);
    if (keySize < RAS_KEY_SIZE) keySize = RAS_KEY_SIZE;
    Pair<RSAPublicKey, RSAPrivateKey> pair = RSAUtils.genKeyPair(keySize);
    //生成公钥和私钥
    RSAPublicKey publicKey = pair.key;
    RSAPrivateKey privateKey = pair.value;

    System.out.println("key generate success!");

    System.out.println("privateKey=" + RSAUtils.encodeBase64(privateKey));
    System.out.println("publicKey=" + RSAUtils.encodeBase64(publicKey));

    //明文
    byte[] ming = "这是一段测试文字。。。。".getBytes(Constants.UTF_8);
    System.out.println("明文:" + new String(ming, Constants.UTF_8));

    //加密后的密文
    byte[] mi = RSAUtils.encryptByPublicKey(ming, publicKey);
    System.out.println("密文:" + new String(mi, Constants.UTF_8));
    //解密后的明文
    ming = RSAUtils.decryptByPrivateKey(mi, privateKey);
    System.out.println("解密:" + new String(ming, Constants.UTF_8));
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
public static byte[] compress(byte[] data) {

        Profiler.enter("time cost on [compress]");

        ByteArrayOutputStream out = new ByteArrayOutputStream(data.length / 4);
        DeflaterOutputStream zipOut = new DeflaterOutputStream(out);
        try {
            zipOut.write(data);
            zipOut.finish();
            zipOut.close();
        } catch (IOException e) {
            LOGGER.error("compress ex", e);
            return Constants.EMPTY_BYTES;
        } finally {
            close(zipOut);
            Profiler.release();
        }
        return out.toByteArray();
    }
 

import com.mpush.api.Constants; //导入依赖的package包/类
@Override
public void saveSession(String sessionContext) {
    File file = new File(rootDir, fileName);
    FileOutputStream out = null;
    try {
        if (!file.exists()) file.getParentFile().mkdirs();
        else if (file.canWrite()) file.delete();
        out = new FileOutputStream(file);
        out.write(sessionContext.getBytes(Constants.UTF_8));
    } catch (Exception e) {
        ClientConfig.I.getLogger().e(e, "save session context ex, session=%s, rootDir=%s"
                , sessionContext, rootDir);
    } finally {
        IOUtils.close(out);
    }
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
@Override
public String getSession() {
    File file = new File(rootDir, fileName);
    if (!file.exists()) return null;
    InputStream in = null;
    try {
        in = new FileInputStream(file);
        byte[] bytes = new byte[in.available()];
        if (bytes.length > 0) {
            in.read(bytes);
            return new String(bytes, Constants.UTF_8);
        }
        in.close();
    } catch (Exception e) {
        ClientConfig.I.getLogger().e(e, "get session context ex,rootDir=%s", rootDir);
    } finally {
        IOUtils.close(in);
    }
    return null;
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
@Override
    public void onReceive(Context context, Intent intent) {

        if (MPushService.ACTION_MESSAGE_RECEIVED.equals(intent.getAction())) {
            byte[] bytes = intent.getByteArrayExtra(MPushService.EXTRA_PUSH_MESSAGE);
            int messageId = intent.getIntExtra(MPushService.EXTRA_PUSH_MESSAGE_ID, 0);
            String message = new String(bytes, Constants.UTF_8);

//           Toast.makeText(context, "收到新的通知：" + message, Toast.LENGTH_SHORT).show();

            if (messageId > 0) MPush.I.ack(messageId);
            if (TextUtils.isEmpty(message)) return;
            reactContext.getJSModule(RCTNativeAppEventEmitter.class).emit(MPUSH_EVENT_MESSAGE, message);

        } else {
            //reactContext.getJSModule(RCTNativeAppEventEmitter.class).emit(MPUSH_EVENT, intent.getAction());
        }
        Log.d("MPush", intent.toString());
        if (MPushService.ACTION_KICK_USER.equals(intent.getAction())) {

            Toast.makeText(context, "用户被踢下线了", Toast.LENGTH_SHORT).show();
        } else if (MPushService.ACTION_BIND_USER.equals(intent.getAction())) {

            Toast.makeText(context, "绑定用户:"
                            + intent.getStringExtra(MPushService.EXTRA_USER_ID)
                            + (intent.getBooleanExtra(MPushService.EXTRA_BIND_RET, false) ? "成功" : "失败")
                    , Toast.LENGTH_SHORT).show();

        } else if (MPushService.ACTION_UNBIND_USER.equals(intent.getAction())) {
            Toast.makeText(context, "解绑用户:"
                            + (intent.getBooleanExtra(MPushService.EXTRA_BIND_RET, false)
                            ? "成功"
                            : "失败")
                    , Toast.LENGTH_SHORT).show();

        } else if (MPushService.ACTION_CONNECTIVITY_CHANGE.equals(intent.getAction())) {
            Toast.makeText(context, intent.getBooleanExtra(MPushService.EXTRA_CONNECT_STATE, false)
                            ? "MPUSH连接建立成功"
                            : "MPUSH连接断开"
                    , Toast.LENGTH_SHORT).show();
        } else if (MPushService.ACTION_HANDSHAKE_OK.equals(intent.getAction())) {
            Toast.makeText(context, "MPUSH握手成功, 心跳:" + intent.getIntExtra(MPushService.EXTRA_HEARTBEAT, 0)
                    , Toast.LENGTH_SHORT).show();
        }
    }
 

import com.mpush.api.Constants; //导入依赖的package包/类
@Override
public void decodeJsonBody(Map<String, Object> body) {
    String content = (String) body.get("content");
    if (content != null) {
        this.content = content.getBytes(Constants.UTF_8);
    }
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
@Override
public Map<String, Object> encodeJsonBody() {
    if (content != null) {
        return Collections.singletonMap("content", new String(content, Constants.UTF_8));
    }
    return null;
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
/**
 * 广播踢人消息到消息中心（MQ）.
 * <p>
 * 有可能目标机器是当前机器，所以要做一次过滤
 * 如果client连续2次链接到同一台机器上就有会出现这中情况
 *
 * @param userId       当前用户
 * @param remoteRouter 用户的路由信息
 */
private void sendKickUserMessage2MQ(String userId, RemoteRouter remoteRouter) {
    ClientLocation location = remoteRouter.getRouteValue();
    //1.如果目标机器是当前机器，就不要再发送广播了，直接忽略
    if (mPushServer.isTargetMachine(location.getHost(), location.getPort())) {
        Logs.CONN.debug("kick remote router in local pc, ignore remote broadcast, userId={}", userId);
        return;
    }

    if (udpGateway) {
        Connection connection = mPushServer.getUdpGatewayServer().getConnection();
        GatewayKickUserMessage.build(connection)
                .setUserId(userId)
                .setClientType(location.getClientType())
                .setConnId(location.getConnId())
                .setDeviceId(location.getDeviceId())
                .setTargetServer(location.getHost())
                .setTargetPort(location.getPort())
                .setRecipient(new InetSocketAddress(location.getHost(), location.getPort()))
                .sendRaw();
    } else {
        //2.发送广播
        //TODO 远程机器可能不存在，需要确认下redis 那个通道如果机器不存在的话，是否会存在消息积压的问题。
        MQKickRemoteMsg message = new MQKickRemoteMsg()
                .setUserId(userId)
                .setClientType(location.getClientType())
                .setConnId(location.getConnId())
                .setDeviceId(location.getDeviceId())
                .setTargetServer(location.getHost())
                .setTargetPort(location.getPort());
        mqClient.publish(Constants.getKickChannel(location.getHostAndPort()), message);
    }
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
/**
 * 获取数据,先从本地获取，本地找不到，从远程获取
 *
 * @param key
 * @return
 */
public String get(final String key) {
    if (null == cache) {
        return null;
    }
    ChildData data = cache.getCurrentData(key);
    if (null != data) {
        return null == data.getData() ? null : new String(data.getData(), Constants.UTF_8);
    }
    return getFromRemote(key);
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
/**
 * 从远程获取数据
 *
 * @param key
 * @return
 */
public String getFromRemote(final String key) {
    if (isExisted(key)) {
        try {
            return new String(client.getData().forPath(key), Constants.UTF_8);
        } catch (Exception ex) {
            Logs.RSD.error("getFromRemote:{}", key, ex);
        }
    }
    return null;
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
/**
 * 更新数据
 *
 * @param key
 * @param value
 */
public void update(final String key, final String value) {
    try {
        /*TransactionOp op = client.transactionOp();
        client.transaction().forOperations(
                op.check().forPath(key),
                op.setData().forPath(key, value.getBytes(Constants.UTF_8))
        );*/
        client.inTransaction().check().forPath(key).and().setData().forPath(key, value.getBytes(Constants.UTF_8)).and().commit();
    } catch (Exception ex) {
        Logs.RSD.error("update:{},{}", key, value, ex);
        throw new ZKException(ex);
    }
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
/**
 * 注册临时数据
 *
 * @param key
 * @param value
 */
public void registerEphemeral(final String key, final String value, boolean cacheNode) {
    try {
        if (isExisted(key)) {
            client.delete().deletingChildrenIfNeeded().forPath(key);
        }
        client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(key, value.getBytes(Constants.UTF_8));
        if (cacheNode) ephemeralNodes.put(key, value);
    } catch (Exception ex) {
        Logs.RSD.error("persistEphemeral:{},{}", key, value, ex);
        throw new ZKException(ex);
    }
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
private int parseTimeout() {
    String timeout = request.headers().get(Constants.HTTP_HEAD_READ_TIMEOUT);
    if (timeout != null) {
        request.headers().remove(Constants.HTTP_HEAD_READ_TIMEOUT);
        Integer integer = Ints.tryParse(timeout);
        if (integer != null && integer > 0) {
            return integer;
        }
    }
    return TIMEOUT;
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
public static String encrypt(String text) {
    try {
        MessageDigest digest = MessageDigest.getInstance("MD5");
        digest.update(text.getBytes(Constants.UTF_8));
        return toHex(digest.digest());
    } catch (Exception e) {
        return Strings.EMPTY;
    }
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
public static String hmacSha1(String data, String encryptKey) {
    final String HMAC_SHA1 = "HmacSHA1";
    SecretKeySpec signingKey = new SecretKeySpec(encryptKey.getBytes(Constants.UTF_8), HMAC_SHA1);
    try {
        Mac mac = Mac.getInstance(HMAC_SHA1);
        mac.init(signingKey);
        mac.update(data.getBytes(Constants.UTF_8));
        return toHex(mac.doFinal());
    } catch (Exception e) {
        return Strings.EMPTY;
    }
}
 

import com.mpush.api.Constants; //导入依赖的package包/类
public static String sha1(String data) {
    try {
        MessageDigest digest = MessageDigest.getInstance("SHA-1");
        return toHex(digest.digest(data.getBytes(Constants.UTF_8)));
    } catch (Exception e) {
        return Strings.EMPTY;
    }
}